-
Notifications
You must be signed in to change notification settings - Fork 377
Expand file tree
/
Copy pathtest_db_containers.py
More file actions
102 lines (87 loc) · 3.72 KB
/
Copy pathtest_db_containers.py
File metadata and controls
102 lines (87 loc) · 3.72 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
import pytest
import sqlalchemy
from pymongo import MongoClient
from testcontainers.core.generic import GenericContainer
from testcontainers.core.waiting_utils import wait_for
from testcontainers.mysql import MySqlContainer, MariaDbContainer
from testcontainers.oracle import OracleDbContainer
from testcontainers.postgres import PostgresContainer
from testcontainers.mongodb import MongoDbContainer
def test_docker_run_mysql():
config = MySqlContainer('mysql:5.7.17')
with config as mysql:
e = sqlalchemy.create_engine(mysql.get_connection_url())
result = e.execute("select version()")
for row in result:
assert row[0] == '5.7.17'
def test_docker_run_postgress():
postgres_container = PostgresContainer("postgres:9.5")
with postgres_container as postgres:
e = sqlalchemy.create_engine(postgres.get_connection_url())
result = e.execute("select version()")
for row in result:
print("server version:", row[0])
def test_docker_run_mariadb():
mariadb_container = MariaDbContainer("mariadb:10.2.9")
with mariadb_container as mariadb:
e = sqlalchemy.create_engine(mariadb.get_connection_url())
result = e.execute("select version()")
for row in result:
assert row[0] == '10.2.9-MariaDB-10.2.9+maria~jessie'
@pytest.mark.skip(reason="needs oracle client libraries unavailable on Travis")
def test_docker_run_oracle():
oracledb_container = OracleDbContainer()
with oracledb_container as oracledb:
e = sqlalchemy.create_engine(oracledb.get_connection_url())
result = e.execute("select * from V$VERSION")
versions = {'Oracle Database 11g Express Edition Release 11.2.0.2.0 - 64bit Production',
'PL/SQL Release 11.2.0.2.0 - Production',
'CORE\t11.2.0.2.0\tProduction',
'TNS for Linux: Version 11.2.0.2.0 - Production',
'NLSRTL Version 11.2.0.2.0 - Production'}
assert {row[0] for row in result} == versions
def test_docker_run_mongodb():
mongo_container = MongoDbContainer("mongo:latest")
with mongo_container as mongo:
db = mongo.get_connection_client().test
doc = {
"address": {
"street": "2 Avenue",
"zipcode": "10075",
"building": "1480",
"coord": [-73.9557413, 40.7720266]
},
"borough": "Manhattan",
"cuisine": "Italian",
"name": "Vella",
"restaurant_id": "41704620"
}
db.restaurants.insert_one(doc)
cursor = db.restaurants.find({"borough": "Manhattan"})
assert cursor.next()['restaurant_id'] == doc['restaurant_id']
def test_docker_generic_db():
mongo_container = GenericContainer("mongo:latest")
mongo_container.with_bind_ports(27017, 27017)
with mongo_container:
def connect():
return MongoClient("mongodb://{}:{}".format(mongo_container.get_container_host_ip(),
mongo_container.get_exposed_port(27017)))
db = wait_for(connect).primer
result = db.restaurants.insert_one(
{
"address": {
"street": "2 Avenue",
"zipcode": "10075",
"building": "1480",
"coord": [-73.9557413, 40.7720266]
},
"borough": "Manhattan",
"cuisine": "Italian",
"name": "Vella",
"restaurant_id": "41704620"
}
)
print(result.inserted_id)
cursor = db.restaurants.find({"borough": "Manhattan"})
for document in cursor:
print(document)