540 units tests not working as expected

dbt/include/sqlserver/macros/materializations/unit_tests.sql

@@ -0,0 +1,47 @@
+{% macro sqlserver__get_unit_test_sql(main_sql, expected_fixture_sql, expected_column_names) -%}
+
+USE [{{ target.database }}];
+IF NOT EXISTS (SELECT * FROM sys.schemas WHERE name = '{{ target.schema }}')
+BEGIN
+EXEC('CREATE SCHEMA [{{ target.schema }}]')
+END
+
+{% set test_view %}
+    [{{ target.schema }}.testview_{{ range(1300, 19000) | random }}]
+{% endset %}
+{% set test_sql = main_sql.replace("'", "''")%}
+EXEC('create view {{test_view}} as {{ test_sql }};')
+
+{% set expected_view %}
+    [{{ target.schema }}.expectedview_{{ range(1300, 19000) | random }}]
+{% endset %}
+{% set expected_sql = expected_fixture_sql.replace("'", "''")%}
+EXEC('create view {{expected_view}} as {{ expected_sql }};')
+
+-- Build actual result given inputs
+{% set unittest_sql %}
+with dbt_internal_unit_test_actual as (
+  select
+    {% for expected_column_name in expected_column_names %}{{expected_column_name}}{% if not loop.last -%},{% endif %}{%- endfor -%}, {{ dbt.string_literal("actual") }} as {{ adapter.quote("actual_or_expected") }}
+  from
+    {{ test_view }}
+),
+-- Build expected result
+dbt_internal_unit_test_expected as (
+  select
+    {% for expected_column_name in expected_column_names %}{{expected_column_name}}{% if not loop.last -%}, {% endif %}{%- endfor -%}, {{ dbt.string_literal("expected") }} as {{ adapter.quote("actual_or_expected") }}
+  from
+    {{ expected_view }}
+)
+-- Union actual and expected results
+select * from dbt_internal_unit_test_actual
+union all
+select * from dbt_internal_unit_test_expected
+{% endset %}
+
+EXEC('{{- escape_single_quotes(unittest_sql) -}}')
+
+EXEC('drop view {{test_view}};')
+EXEC('drop view {{expected_view}};')
+
+{%- endmacro %}

dbt/include/sqlserver/macros/readme.md

@@ -48,3 +48,7 @@ DBT expects a limit function, but the sqlserver syntax does not support it. Fabr
 ## `sqlserver__snapshot_merge_sql`
 
 Restores logic to the merge statement logic like the dbt core. Merge will probably be slower then the existing logic
+
+## unit tests
+
+To accomidate the nested CTE situation, we create a temp view for the actual/expected and use those both in the test.

tests/functional/adapter/dbt/test_unit_tests.py

@@ -0,0 +1,75 @@
+import pytest
+from dbt.tests.adapter.unit_testing.test_case_insensitivity import BaseUnitTestCaseInsensivity
+from dbt.tests.adapter.unit_testing.test_invalid_input import BaseUnitTestInvalidInput
+from dbt.tests.adapter.unit_testing.test_types import BaseUnitTestingTypes
+from dbt.tests.util import run_dbt, write_file
+
+my_model_sql = """
+select
+    tested_column from {{ ref('my_upstream_model')}}
+"""
+
+my_upstream_model_sql = """
+select
+  {sql_value} as tested_column
+"""
+
+test_my_model_yml = """
+unit_tests:
+  - name: test_my_model
+    model: my_model
+    given:
+      - input: ref('my_upstream_model')
+        rows:
+          - {{ tested_column: {yaml_value} }}
+    expect:
+      rows:
+        - {{ tested_column: {yaml_value} }}
+"""
+
+
+class TestUnitTestCaseInsensitivity(BaseUnitTestCaseInsensivity):
+    pass
+
+
+class TestUnitTestInvalidInput(BaseUnitTestInvalidInput):
+    pass
+
+
+class TestUnitTestingTypes(BaseUnitTestingTypes):
+    @pytest.fixture
+    def data_types(self):
+        # sql_value, yaml_value
+        return [
+            ["1", "1"],
+            ["'1'", "1"],
+            ["1", "true"],
+            ["CAST('2020-01-02' AS DATE)", "2020-01-02"],
+            ["CAST('2013-11-03 00:00:00-0' AS DATETIME2(6))", "2013-11-03 00:00:00-0"],
+            ["CAST('2013-11-03 00:00:00-0' AS DATETIME2(6))", "2013-11-03 00:00:00-0"],
+            ["CAST('1' AS numeric)", "1"],
+        ]
+
+    def test_unit_test_data_type(self, project, data_types):
+        for sql_value, yaml_value in data_types:
+            # Write parametrized type value to sql files
+            write_file(
+                my_upstream_model_sql.format(sql_value=sql_value),
+                "models",
+                "my_upstream_model.sql",
+            )
+
+            # Write parametrized type value to unit test yaml definition
+            write_file(
+                test_my_model_yml.format(yaml_value=yaml_value),
+                "models",
+                "schema.yml",
+            )
+
+            results = run_dbt(["run", "--select", "my_upstream_model"])
+            assert len(results) == 1
+
+            try:
+                run_dbt(["test", "--select", "my_model"])
+            except Exception:
+                raise AssertionError(f"unit test failed when testing model with {sql_value}")