-
Notifications
You must be signed in to change notification settings - Fork 94
/
Copy pathtest_sources.py
69 lines (60 loc) · 2.12 KB
/
test_sources.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
import os
import pytest
from dbt.tests.util import run_dbt
sources_schema_yml = """version: 2
sources:
- name: external_source
config:
external_location: "/tmp/{name}_{extra}.csv"
tables:
- name: seeds_source
description: "A source table"
config:
extra: 'something'
columns:
- name: id
description: "An id"
tests:
- unique
- not_null
- name: seeds_ost
identifier: "seeds_other_source_table"
config:
external_location: "read_csv_auto('/tmp/%(identifier)s.csv')"
formatter: oldstyle
- name: seeds_other_source_table
config:
external_location: "read_csv_auto('/tmp/${name}.csv')"
formatter: template
"""
models_source_model_sql = """select * from {{ source('external_source', 'seeds_source') }}
"""
models_multi_source_model_sql = """select s.* from {{ source('external_source', 'seeds_source') }} s
inner join {{ source('external_source', 'seeds_ost') }} oldstyle USING (id)
inner join {{ source('external_source', 'seeds_other_source_table') }} tmpl USING (id)
"""
class TestExternalSources:
@pytest.fixture(scope="class")
def models(self):
return {
"schema.yml": sources_schema_yml,
"source_model.sql": models_source_model_sql,
"multi_source_model.sql": models_multi_source_model_sql,
}
@pytest.fixture(scope="class")
def seeds_source_file(self):
with open("/tmp/seeds_source_something.csv", "w") as f:
f.write("id,a,b\n1,2,3\n4,5,6\n7,8,9")
yield
os.unlink("/tmp/seeds_source_something.csv")
@pytest.fixture(scope="class")
def ost_file(self):
with open("/tmp/seeds_other_source_table.csv", "w") as f:
f.write("id,c,d\n1,2,3\n4,5,6\n7,8,9")
yield
os.unlink("/tmp/seeds_other_source_table.csv")
def test_external_sources(self, seeds_source_file, ost_file, project):
results = run_dbt(["run"])
assert len(results) == 2
test_results = run_dbt(["test"])
assert len(test_results) == 2