tests: api: separate live tests

tests/func/test_api.py → tests/func/api/test_api.py

@@ -1,16 +1,10 @@
-import csv
-import io
-import json
-import logging
 import os
-from typing import Dict, List
 
 import pytest
 from flaky.flaky_decorator import flaky
-from funcy import first, get_in, last
+from funcy import first, get_in
 
 from dvc import api
-from dvc.api.live import summary
 from dvc.exceptions import FileMissingError, OutputNotFoundError
 from dvc.path_info import URLInfo
 from dvc.utils.fs import remove
@@ -277,54 +271,3 @@ def test_get_url_subrepos(tmp_dir, scm, local_cloud):
 
     expected_url = os.path.join(path, "37", "b51d194a7513e45b56f6524f2d51f2")
     assert api.get_url("subrepo/bar") == expected_url
-
-
-def _dumps_sv(metrics: List[Dict], delimiter=","):
-    stream = io.StringIO()
-    writer = csv.DictWriter(
-        stream, fieldnames=list(first(metrics).keys()), delimiter=delimiter
-    )
-    writer.writeheader()
-    writer.writerows(metrics)
-    stream.seek(0)
-    return stream.read()
-
-
-@pytest.fixture
-def live_results(tmp_dir):
-    def make(path="logs"):
-        datapoints = [{"metric": 0.0, "step": 0}, {"metric": 0.5, "step": 1}]
-        tmp_dir.gen(
-            {
-                (tmp_dir / path).with_suffix(".json"): json.dumps(
-                    last(datapoints)
-                ),
-                (tmp_dir / path / "metric.tsv"): _dumps_sv(
-                    datapoints, delimiter="\t",
-                ),
-            }
-        )
-
-    yield make
-
-
-def test_live_summary_no_repo(tmp_dir, live_results, caplog):
-    live_results("logs")
-
-    with caplog.at_level(logging.INFO, logger="dvc"):
-        summary("logs")
-
-    summary_path = tmp_dir / "logs.html"
-    assert summary_path.exists()
-    assert f"file://{str(summary_path)}" in caplog.text
-
-
-def test_live_summary(tmp_dir, dvc, live_results, caplog):
-    live_results("logs")
-
-    with caplog.at_level(logging.INFO, logger="dvc"):
-        summary("logs")
-
-    summary_path = tmp_dir / "logs.html"
-    assert summary_path.exists()
-    assert f"file://{str(summary_path)}" in caplog.text

tests/func/api/test_live.py

@@ -0,0 +1,55 @@
+import io
+import json
+import logging
+from typing import Dict, List
+
+import pytest
+from funcy import last
+
+from dvc.api.live import summary
+from tests.utils import dump_sv
+
+
+def _dumps_tsv(metrics: List[Dict]):
+    stream = io.StringIO()
+    dump_sv(stream, metrics)
+    stream.seek(0)
+    return stream.read()
+
+
+@pytest.fixture
+def live_results(tmp_dir):
+    def make(path="logs"):
+        datapoints = [{"metric": 0.0, "step": 0}, {"metric": 0.5, "step": 1}]
+        tmp_dir.gen(
+            {
+                (tmp_dir / path).with_suffix(".json"): json.dumps(
+                    last(datapoints)
+                ),
+                (tmp_dir / path / "metric.tsv"): _dumps_tsv(datapoints),
+            }
+        )
+
+    yield make
+
+
+def test_live_summary_no_repo(tmp_dir, live_results, caplog):
+    live_results("logs")
+
+    with caplog.at_level(logging.INFO, logger="dvc"):
+        summary("logs")
+
+    summary_path = tmp_dir / "logs.html"
+    assert summary_path.exists()
+    assert f"file://{str(summary_path)}" in caplog.text
+
+
+def test_live_summary(tmp_dir, dvc, live_results, caplog):
+    live_results("logs")
+
+    with caplog.at_level(logging.INFO, logger="dvc"):
+        summary("logs")
+
+    summary_path = tmp_dir / "logs.html"
+    assert summary_path.exists()
+    assert f"file://{str(summary_path)}" in caplog.text

tests/func/plots/utils.py

@@ -1,22 +1,11 @@
-import csv
 import json
 
-from funcy import first
+from tests.utils import dump_sv
 
 
 def _write_csv(metric, filename, header=True):
     with open(filename, "w", newline="") as csvobj:
-        if header:
-            writer = csv.DictWriter(
-                csvobj, fieldnames=list(first(metric).keys())
-            )
-            writer.writeheader()
-            writer.writerows(metric)
-        else:
-            writer = csv.writer(csvobj)
-            for d in metric:
-                assert len(d) == 1
-                writer.writerow(list(d.values()))
+        dump_sv(csvobj, metric, delimiter=",", header=header)
 
 
 def _write_json(tmp_dir, metric, filename):

tests/utils/__init__.py

@@ -1,6 +1,9 @@
+import csv
 import os
 from contextlib import contextmanager
 
+from funcy import first
+
 from dvc.scm import Git
 
 
@@ -21,3 +24,16 @@ def cd(newdir):
 
 def to_posixpath(path):
     return path.replace("\\", "/")
+
+
+def dump_sv(stream, metrics, delimiter=",", header=True):
+    if header:
+        writer = csv.DictWriter(
+            stream, fieldnames=list(first(metrics).keys()), delimiter=delimiter
+        )
+        writer.writeheader()
+        writer.writerows(metrics)
+    else:
+        writer = csv.writer(stream)
+        for d in metrics:
+            writer.writerow(list(d.values()))