-
Notifications
You must be signed in to change notification settings - Fork 104
/
Copy pathtest_test_sqlserver.py
77 lines (57 loc) · 2.4 KB
/
test_test_sqlserver.py
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
import os
import pytest
from testcontainers.mssql import SqlServerContainer
from datacontract.data_contract import DataContract
# logging.basicConfig(level=logging.DEBUG, force=True)
datacontract = "fixtures/sqlserver/datacontract.yaml"
sql_file_path = "fixtures/sqlserver/data/data.sql"
sql_server = SqlServerContainer()
SQL_SERVER_PORT: int = 1433
@pytest.fixture(scope="module", autouse=True)
def mssql_container(request):
sql_server.start()
def remove_container():
sql_server.stop()
request.addfinalizer(remove_container)
# To run this on ARM macs, run these commands (https://github.com/pymssql/pymssql/issues/769)
# brew install FreeTDS
# export LDFLAGS="-L/opt/homebrew/lib -L/opt/homebrew/opt/openssl/lib"
# export CFLAGS="-I/opt/homebrew/include"
# export CPPFLAGS="-I/opt/homebrew/opt/openssl@3/include"
# pip uninstall pymssql -y
# pip install pymssql==2.2.8 --no-binary :all:
@pytest.mark.skipif(not os.getenv("CI"), reason="Skipping test outside CI/CD environment")
def test_test_sqlserver(mssql_container, monkeypatch):
monkeypatch.setenv("DATACONTRACT_SQLSERVER_USERNAME", sql_server.username)
monkeypatch.setenv("DATACONTRACT_SQLSERVER_PASSWORD", sql_server.password)
monkeypatch.setenv("DATACONTRACT_SQLSERVER_TRUST_SERVER_CERTIFICATE", "True")
_init_sql()
data_contract_str = _setup_datacontract()
data_contract = DataContract(data_contract_str=data_contract_str)
run = data_contract.test()
print(run)
assert run.result == "passed"
assert all(check.result == "passed" for check in run.checks)
def _setup_datacontract():
with open(datacontract) as data_contract_file:
data_contract_str = data_contract_file.read()
port = sql_server.get_exposed_port(SQL_SERVER_PORT)
data_contract_str = data_contract_str.replace("__PORT__", port)
return data_contract_str
def _init_sql():
# import locally as a top level import of pymssql fails on ARM macs
import pymssql
connection = pymssql.connect(
database=sql_server.dbname,
user=sql_server.username,
password=sql_server.password,
host=sql_server.get_container_host_ip(),
port=sql_server.get_exposed_port(SQL_SERVER_PORT),
)
cursor = connection.cursor()
with open(sql_file_path, "r") as sql_file:
sql_commands = sql_file.read()
cursor.execute(sql_commands)
connection.commit()
cursor.close()
connection.close()