API response recorder for test fixtures

requirements-test.txt

@@ -7,5 +7,6 @@ pylint>=3.0.2
 mypy>=1.7.1
 molot~=1.0.0
 dbt-postgres~=1.8.1
+python-dotenv~=1.0.1
 types-requests
 types-PyYAML

tests/_mocks.py

@@ -1,8 +1,10 @@
 import json
+import os
 from pathlib import Path
-from typing import Any, Dict, Mapping, Optional, Sequence
+from typing import Any, Dict, Mapping, Optional, Sequence, Union
 
 import requests
+from dotenv import dotenv_values
 
 from dbtmetabase.core import DbtMetabase
 from dbtmetabase.manifest import Manifest, Model
@@ -11,14 +13,28 @@
 FIXTURES_PATH = Path("tests") / "fixtures"
 TMP_PATH = Path("tests") / "tmp"
 
+RECORD = os.getenv("RECORD", "").lower() == "true"
+
 
 class MockMetabase(Metabase):
-    def __init__(self, url: str):
+    def __init__(self, url: str, record: bool = False):
+        self.record = record
+
+        api_key = "dummy"
+        username = None
+        password = None
+
+        if record:
+            env = dotenv_values(Path().parent / "sandbox" / ".env")
+            api_key = None
+            username = env["MB_USER"]
+            password = env["MB_PASSWORD"]
+
         super().__init__(
             url=url,
-            api_key="dummy",
-            username="None",
-            password=None,
+            api_key=api_key,
+            username=username,
+            password=password,
             session_id=None,
             skip_verify=False,
             cert=None,
@@ -33,18 +49,29 @@ def _api(
         path: str,
         params: Optional[Dict[str, Any]] = None,
         **kwargs,
-    ) -> Mapping:
+    ) -> Union[Mapping, Sequence]:
         path_toks = f"{path.lstrip('/')}.json".split("/")
-        if path_toks[0] == "api" and method == "get":
-            json_path = Path.joinpath(FIXTURES_PATH, *path_toks)
-            if json_path.exists():
-                with open(json_path, encoding="utf-8") as f:
-                    return json.load(f)
-            else:
-                response = requests.Response()
-                response.status_code = 404
-                raise requests.exceptions.HTTPError(response=response)
-        return {}
+        json_path = Path.joinpath(FIXTURES_PATH, *path_toks)
+
+        if self.record:
+            resp = super()._api(method, path, params, **kwargs)
+
+            if method == "get":
+                json_path.parent.mkdir(parents=True, exist_ok=True)
+                with open(json_path, "w", encoding="utf-8") as f:
+                    json.dump(resp, f, indent=4)
+
+            return resp
+        else:
+            if method == "get":
+                if json_path.exists():
+                    with open(json_path, encoding="utf-8") as f:
+                        return json.load(f)
+                else:
+                    response = requests.Response()
+                    response.status_code = 404
+                    raise requests.exceptions.HTTPError(response=response)
+            return {}
 
 
 class MockManifest(Manifest):
@@ -63,4 +90,4 @@ def __init__(
         metabase_url: str = "http://localhost:3000",
     ):  # pylint: disable=super-init-not-called
         self._manifest = MockManifest(path=manifest_path)
-        self._metabase = MockMetabase(url=metabase_url)
+        self._metabase = MockMetabase(url=metabase_url, record=RECORD)