2
2
from pymongo import MongoClient
3
3
from pymongo .errors import OperationFailure
4
4
5
- from testcontainers .core .container import DockerContainer
6
- from testcontainers .core .waiting_utils import wait_for
7
5
from testcontainers .mongodb import MongoDbContainer
8
6
9
7
10
- def test_docker_generic_db ():
11
- with DockerContainer ("mongo:latest" ).with_bind_ports (27017 , 27017 ) as mongo_container :
12
-
13
- def connect ():
14
- host = mongo_container .get_container_host_ip ()
15
- port = mongo_container .get_exposed_port (27017 )
16
- return MongoClient (f"mongodb://{ host } :{ port } " )
17
-
18
- db = wait_for (connect ).primer
19
- result = db .restaurants .insert_one (
20
- {
21
- "address" : {
22
- "street" : "2 Avenue" ,
23
- "zipcode" : "10075" ,
24
- "building" : "1480" ,
25
- "coord" : [- 73.9557413 , 40.7720266 ],
26
- },
27
- "borough" : "Manhattan" ,
28
- "cuisine" : "Italian" ,
29
- "name" : "Vella" ,
30
- "restaurant_id" : "41704620" ,
31
- }
32
- )
33
- assert result .inserted_id
34
- cursor = db .restaurants .find ({"borough" : "Manhattan" })
35
- for document in cursor :
36
- assert document
37
-
38
-
39
- def test_docker_run_mongodb ():
40
- with MongoDbContainer ("mongo:latest" ) as mongo :
8
+ @pytest .mark .parametrize ("version" , ["7.0.7" , "6.0.14" , "5.0.26" ])
9
+ def test_docker_run_mongodb (version : str ):
10
+ with MongoDbContainer (f"mongo:{ version } " ) as mongo :
41
11
db = mongo .get_connection_client ().test
42
12
doc = {
43
13
"address" : {
@@ -51,14 +21,8 @@ def test_docker_run_mongodb():
51
21
"name" : "Vella" ,
52
22
"restaurant_id" : "41704620" ,
53
23
}
54
- db .restaurants .insert_one (doc )
24
+ result = db .restaurants .insert_one (doc )
25
+ assert result .inserted_id
26
+
55
27
cursor = db .restaurants .find ({"borough" : "Manhattan" })
56
28
assert cursor .next ()["restaurant_id" ] == doc ["restaurant_id" ]
57
-
58
-
59
- def test_docker_run_mongodb_connect_without_credentials ():
60
- with MongoDbContainer () as mongo :
61
- connection_url = f"mongodb://{ mongo .get_container_host_ip ()} :" f"{ mongo .get_exposed_port (mongo .port )} "
62
- db = MongoClient (connection_url ).test
63
- with pytest .raises (OperationFailure ):
64
- db .restaurants .insert_one ({})
0 commit comments