datacontract-cli/tests/test_import_sql_postgres.py at f569c9fa5f23fc414d3600af48e8fe341acaaa94 · dmaresma/datacontract-cli · GitHub

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
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
import yaml
from typer.testing import CliRunner

from datacontract.cli import app
from datacontract.data_contract import DataContract

# logging.basicConfig(level=logging.DEBUG, force=True)

datacontract = "fixtures/postgres/datacontract.yaml"
sql_file_path = "fixtures/postgres/data/data.sql"


def test_cli():
    runner = CliRunner()
    result = runner.invoke(
        app,
        [
            "import",
            "--format",
            "sql",
            "--source",
            sql_file_path,
            "--dialect",
            "postgres"
        ],
    )
    assert result.exit_code == 0


def test_import_sql_postgres():
    result = DataContract().import_from_source("sql", sql_file_path, dialect="postgres")

    expected = """
dataContractSpecification: 1.2.0
id: my-data-contract-id
info:
  title: My Data Contract
  version: 0.0.1
servers:
  postgres:
    type: postgres
models:
  my_table:
    type: table
    fields:
      field_one:
        type: string
        primaryKey: true
        maxLength: 10
        config:
          postgresType: VARCHAR(10)
      field_two:
        type: int
        required: true
        config:
          postgresType: INT
      field_three:
        type: timestamp_tz
        config:
          postgresType: TIMESTAMPTZ
    """
    print("Result", result.to_yaml())
    assert yaml.safe_load(result.to_yaml()) == yaml.safe_load(expected)
    # Disable linters so we don't get "missing description" warnings
    assert DataContract(data_contract_str=expected).lint(enabled_linters=set()).has_passed()


def test_import_sql_constraints():
    result = DataContract().import_from_source("sql", "fixtures/postgres/data/data_constraints.sql", dialect="postgres")

    expected = """
dataContractSpecification: 1.2.0
id: my-data-contract-id
info:
  title: My Data Contract
  version: 0.0.1
servers:
  postgres:
    type: postgres
models:
  customer_location:
    type: table
    fields:
      id:
        type: decimal
        required: true
        # primaryKey: true
        config:
          postgresType: DECIMAL
      created_by:
        type: string
        required: true
        maxLength: 30
        config:
          postgresType: VARCHAR(30)
      create_date:
        type: timestamp_ntz
        required: true
        config:
          postgresType: TIMESTAMP
      changed_by:
        type: string
        maxLength: 30
        config:
          postgresType: VARCHAR(30)
      change_date:
        type: timestamp_ntz
        config:
          postgresType: TIMESTAMP
      name:
        type: string
        required: true
        maxLength: 120
        config:
          postgresType: VARCHAR(120)
      short_name:
        type: string
        maxLength: 60
        config:
          postgresType: VARCHAR(60)
      display_name:
        type: string
        required: true
        maxLength: 120
        config:
          postgresType: VARCHAR(120)
      code:
        type: string
        required: true
        maxLength: 30
        config:
          postgresType: VARCHAR(30)
      description:
        type: string
        maxLength: 4000
        config:
          postgresType: VARCHAR(4000)
      language_id:
        type: decimal
        required: true
        config:
          postgresType: DECIMAL
      status:
        type: string
        required: true
        maxLength: 2
        config:
          postgresType: VARCHAR(2)
    """
    print("Result", result.to_yaml())
    assert yaml.safe_load(result.to_yaml()) == yaml.safe_load(expected)
    # Disable linters so we don't get "missing description" warnings
    assert DataContract(data_contract_str=expected).lint(enabled_linters=set()).has_passed()