1
- import pytest
1
+ import re
2
+ from pathlib import Path
2
3
from typing import Optional
4
+
5
+ import pytest
6
+ from httpx import get
7
+
8
+ from testcontainers .core .waiting_utils import wait_for_logs
9
+ from testcontainers .core .image import DockerImage
3
10
from testcontainers .core .generic import ServerContainer
4
11
5
- import re
12
+ TEST_DIR = Path ( __file__ ). parent
6
13
7
14
8
15
@pytest .mark .parametrize ("test_image_cleanup" , [True , False ])
9
16
@pytest .mark .parametrize ("test_image_tag" , [None , "custom-image:test" ])
10
17
def test_srv_container (test_image_tag : Optional [str ], test_image_cleanup : bool , check_for_image , port = 9000 ):
11
- with ServerContainer (
12
- path = "./core/tests/image_fixtures/python_server" ,
13
- port = port ,
14
- tag = test_image_tag ,
15
- image_cleanup = test_image_cleanup ,
16
- ) as srv :
17
- image_short_id = srv .docker_image .short_id
18
- image_build_logs = srv .docker_image .get_logs ()
18
+ with (
19
+ DockerImage (
20
+ path = TEST_DIR / "image_fixtures/python_server" ,
21
+ tag = test_image_tag ,
22
+ clean_up = test_image_cleanup ,
23
+ #
24
+ ) as docker_image ,
25
+ ServerContainer (port = port , image = docker_image ) as srv ,
26
+ ):
27
+ image_short_id = docker_image .short_id
28
+ image_build_logs = docker_image .get_logs ()
19
29
# check if dict is in any of the logs
20
30
assert {"stream" : f"Step 2/3 : EXPOSE { port } " } in image_build_logs , "Image logs mismatch"
21
31
assert (port , None ) in srv .ports .items (), "Port mismatch"
@@ -25,3 +35,13 @@ def test_srv_container(test_image_tag: Optional[str], test_image_cleanup: bool,
25
35
assert re .match (r"http://localhost:\d+" , test_url ), "Connection URL mismatch"
26
36
27
37
check_for_image (image_short_id , test_image_cleanup )
38
+
39
+
40
+ def test_like_doctest ():
41
+ with DockerImage (path = TEST_DIR / "image_fixtures/python_server" , tag = "test-srv:latest" ) as image :
42
+ with ServerContainer (port = 9000 , image = image ) as srv :
43
+ url = srv ._create_connection_url ()
44
+ response = get (f"{ url } " , timeout = 5 )
45
+ assert response .status_code == 200 , "Response status code is not 200"
46
+ delay = wait_for_logs (srv , "GET / HTTP/1.1" )
47
+ print (delay )
0 commit comments