Convert test to use logging event

core/dbt/tests/util.py

@@ -116,24 +116,39 @@ def run_dbt(
 # If you want the logs that are normally written to a file, you must
 # start with the "--debug" flag. The structured schema log CI test
 # will turn the logs into json, so you have to be prepared for that.
-def run_dbt_and_capture(args: List[str] = None, expect_pass=True):
+def run_dbt_and_capture(
+    args: List[str] = None,
+    expect_pass: bool = True,
+    publications: List[PublicationArtifact] = None,
+):
     try:
         stringbuf = StringIO()
         capture_stdout_logs(stringbuf)
-        res = run_dbt(args, expect_pass=expect_pass)
+        res = run_dbt(args, expect_pass=expect_pass, publications=publications)
         stdout = stringbuf.getvalue()
 
     finally:
         stop_capture_stdout_logs()
 
-    # Json logs will have lots of escape characters which will
-    # make checks for strings in the logs fail, so remove those.
-    if '{"code":' in stdout:
-        stdout = stdout.replace("\\", "")
-
     return res, stdout
 
 
+def get_logging_events(log_output, event_name):
+    logging_events = []
+    for log_line in log_output.split("\n"):
+        # skip empty lines
+        if len(log_line) == 0:
+            continue
+        # The adapter logging also shows up, so skip non-json lines
+        if not log_line.startswith("{"):
+            continue
+        if event_name in log_line:
+            log_dct = json.loads(log_line)
+            if log_dct["info"]["name"] == event_name:
+                logging_events.append(log_dct)
+    return logging_events
+
+
 # Used in test cases to get the manifest from the partial parsing file
 # Note: this uses an internal version of the manifest, and in the future
 # parts of it will not be supported for external use.

tests/functional/multi_project/test_publication.py

@@ -2,7 +2,12 @@
 import pytest
 import os
 
-from dbt.tests.util import run_dbt, get_artifact, write_file
+from dbt.tests.util import (
+    run_dbt,
+    write_file,
+    run_dbt_and_capture,
+    get_logging_events,
+)
 from dbt.contracts.publication import PublicationArtifact, PublicModel
 from dbt.exceptions import (
     PublicationConfigNotFound,
@@ -102,10 +107,11 @@ def models(self):
         }
 
     def test_publication_artifact(self, project):
-        results = run_dbt(["run"])
+        results, log_output = run_dbt_and_capture(["--debug", "--log-format=json", "run"])
         assert len(results) == 3
 
-        publication_dict = get_artifact(project.project_root, "target", "test_publication.json")
+        pub_available_events = get_logging_events(log_output, "PublicationArtifactAvailable")
+        publication_dict = pub_available_events[0]["data"]["pub_artifact"]
         publication = PublicationArtifact.from_dict(publication_dict)
         assert publication
         assert len(publication.public_models) == 2
@@ -134,13 +140,16 @@ def test_pub_artifacts(self, project):
         # Provide publication and try again
         m_pub_json = marketing_pub_json.replace("test_schema", project.test_schema)
         publications = [PublicationArtifact.from_dict(json.loads(m_pub_json))]
-        manifest = run_dbt(["parse"], publications=publications)
+        manifest, log_output = run_dbt_and_capture(
+            ["--debug", "--log-format=json", "parse"], publications=publications
+        )
         assert manifest.publications
         assert "marketing" in manifest.publications
         assert "model.marketing.fct_one" in manifest.publications["marketing"].public_node_ids
 
         # Check dependencies in publication_artifact
-        publication_dict = get_artifact(project.project_root, "target", "test_publication.json")
+        pub_available_events = get_logging_events(log_output, "PublicationArtifactAvailable")
+        publication_dict = pub_available_events[0]["data"]["pub_artifact"]
         publication = PublicationArtifact.from_dict(publication_dict)
         assert publication.dependencies == ["marketing"]
 
@@ -234,13 +243,14 @@ def test_multi_projects(self, project, project_alt):
         # run the alternate project by using the alternate project root
         # (There is currently a bug where project-dir requires a chdir to work.)
         os.chdir(project_alt.project_root)
-        results = run_dbt(["run", "--project-dir", str(project_alt.project_root)])
+        results, log_output = run_dbt_and_capture(
+            ["--debug", "--log-format=json", "run", "--project-dir", str(project_alt.project_root)]
+        )
         assert len(results) == 1
 
         # Check publication artifact
-        publication_dict = get_artifact(
-            project_alt.project_root, "target", "test_alt_publication.json"
-        )
+        pub_available_events = get_logging_events(log_output, "PublicationArtifactAvailable")
+        publication_dict = pub_available_events[0]["data"]["pub_artifact"]
         publication = PublicationArtifact.from_dict(publication_dict)
         assert len(publication.public_models) == 1