tests: stub out constructor tests

Author: machow

.env.dev

@@ -1,3 +1,7 @@
+# Pins optional config ----
+#PINS_CACHE_DIR=.pins_cache
+#PINS_DATA_DIR=.pins_data
+
 # AWS S3 backend ----
 AWS_ACCESS_KEY_ID=
 AWS_SECRET_ACCESS_KEY=

pins/tests/conftest.py

@@ -1,22 +1,29 @@
 import pytest
 import tempfile
+import os
 
 from importlib_resources import files
 from pytest import mark as m
 from pathlib import Path
-from pins.tests.helpers import BoardBuilder, RscBoardBuilder, Snapshot
+from pins.tests.helpers import BoardBuilder, RscBoardBuilder, Snapshot, rm_env
+
+PATH_TO_EXAMPLE_BOARD = files("pins") / "tests/pins-compat"
+
 
 # Based on https://github.com/machow/siuba/blob/main/siuba/tests/helpers.py
 BACKEND_MARKS = ["fs_s3", "fs_file", "fs_rsc"]
 
-param_rsc = pytest.param(lambda: RscBoardBuilder("rsc"), id="rsc", marks=m.fs_rsc)
-
-params_backend = [
+# parameters that can be used more than once per session
+params_safe = [
     pytest.param(lambda: BoardBuilder("file"), id="file", marks=m.fs_file),
     pytest.param(lambda: BoardBuilder("s3"), id="s3", marks=m.fs_s3),
-    param_rsc,
 ]
 
+# rsc should only be used once, because users are created at docker setup time
+param_rsc = pytest.param(lambda: RscBoardBuilder("rsc"), id="rsc", marks=m.fs_rsc)
+
+params_backend = [*params_safe, param_rsc]
+
 
 @pytest.fixture(params=params_backend, scope="session")
 def backend(request):
@@ -25,6 +32,15 @@ def backend(request):
     backend.teardown()
 
 
+@pytest.fixture(scope="session")
+def http_example_board_path():
+    # backend = BoardBuilder("s3")
+    # yield backend.create_tmp_board(str(PATH_TO_EXAMPLE_BOARD.absolute())).board
+    # backend.teardown()
+    # TODO: could putting it in a publically available bucket folder
+    return "https://raw.githubusercontent.com/machow/pins-python/main/pins/tests/pins-compat"
+
+
 @pytest.fixture
 def snapshot(request):
     p_snap = files("pins") / "tests/_snapshots" / request.node.originalname
@@ -49,6 +65,22 @@ def tmp_dir2():
         yield Path(tmp_dir)
 
 
+@pytest.fixture
+def tmp_cache(tmp_dir2):
+    with rm_env("PINS_CACHE_DIR"):
+        os.environ["PINS_CACHE_DIR"] = str(tmp_dir2)
+
+        yield tmp_dir2
+
+
+@pytest.fixture
+def tmp_data_dir(tmp_dir2):
+    with rm_env("PINS_DATA_DIR"):
+        os.environ["PINS_DATA_DIR"] = str(tmp_dir2)
+
+        yield tmp_dir2
+
+
 def pytest_addoption(parser):
     parser.addoption("--snapshot-update", action="store_true")
 

pins/tests/helpers.py

@@ -1,5 +1,6 @@
 import pytest
 
+import contextlib
 import uuid
 import os
 import json
@@ -104,7 +105,8 @@ def __init__(self, fs_name: str, path=None):
         self.fs_name = fs_name
         self.board_path_registry = []
 
-    def get_param_from_env(self, entry):
+    @staticmethod
+    def get_param_from_env(entry):
         name, default = entry
 
         value = os.environ.get(name, default)
@@ -249,3 +251,22 @@ def assert_equal_file(self, dst_file):
             shutil.copy(dst_file, self.path)
 
         assert self.path.read_text() == Path(dst_file).read_text()
+
+
+@contextlib.contextmanager
+def rm_env(*args):
+    """
+    Temporarily set the process environment variables.
+
+
+    """
+    old_environ = dict(os.environ)
+    for arg in args:
+        if arg in os.environ:
+            del os.environ[arg]
+
+    try:
+        yield
+    finally:
+        os.environ.clear()
+        os.environ.update(old_environ)

pins/tests/test_compat.py

@@ -1,24 +1,21 @@
 import pytest
 import datetime
 
-import importlib_resources as resources
 
 from pins.errors import PinsError
 from pins.tests.helpers import xfail_fs
+from pins.tests.conftest import PATH_TO_EXAMPLE_BOARD
 
 
 NOT_A_PIN = "not_a_pin_abcdefg"
 PIN_CSV = "df_csv"
 
-path_to_board = resources.files("pins") / "tests/pins-compat"
-
-
 # set up board ----
 
 
 @pytest.fixture(scope="session")
 def board(backend):
-    board = backend.create_tmp_board(str(path_to_board.absolute()))
+    board = backend.create_tmp_board(str(PATH_TO_EXAMPLE_BOARD.absolute()))
 
     yield board
 
@@ -112,7 +109,7 @@ def test_compat_pin_meta_version_arg_error(board):
 def test_compat_pin_read(board):
     import pandas as pd
 
-    p_data = path_to_board / "df_csv" / "20220214T163720Z-9bfad" / "df_csv.csv"
+    p_data = PATH_TO_EXAMPLE_BOARD / "df_csv" / "20220214T163720Z-9bfad" / "df_csv.csv"
 
     src_df = board.pin_read("df_csv")
     dst_df = pd.read_csv(p_data, index_col=0)

pins/tests/test_constructors.py

@@ -1,36 +1,66 @@
-import contextlib
 import os
 import pytest
 
 from pathlib import Path
 
 from pins import constructors as c
+from pins.tests.conftest import PATH_TO_EXAMPLE_BOARD
+from pins.tests.helpers import rm_env
+
 
 # adapted from https://stackoverflow.com/a/34333710
 
 
-@contextlib.contextmanager
-def rm_env(*args):
-    """
-    Temporarily set the process environment variables.
+def check_dir_writable(p_dir):
+    assert p_dir.parent.exists()
+    assert os.access(p_dir.parent.absolute(), os.W_OK)
 
 
-    """
-    old_environ = dict(os.environ)
-    for arg in args:
-        if arg in os.environ:
-            del os.environ[arg]
+# End-to-end constructor tests
 
-    try:
-        yield
-    finally:
-        os.environ.clear()
-        os.environ.update(old_environ)
+# there are two facets of boards: reading and writing.
+# copied from test_compat
+def test_constructor_board_url(tmp_cache, http_example_board_path):
+    board = c.board_urls(
+        http_example_board_path, pin_paths={"df_csv": "df_csv/20220214T163718Z-eceac/"}
+    )
 
+    board.pin_read("df_csv")
 
-def check_dir_writable(p_dir):
-    assert p_dir.parent.exists()
-    assert os.access(p_dir.parent.absolute(), os.W_OK)
+    # check cache
+    # check data
+
+
+def test_constructor_board_github(tmp_cache, http_example_board_path):
+    board = c.board_github("machow", "pins-python", PATH_TO_EXAMPLE_BOARD)  # noqa
+
+
+@pytest.fixture(scope="session")
+def board(backend):
+    # TODO: copied from test_compat.py
+
+    board = backend.create_tmp_board(str(PATH_TO_EXAMPLE_BOARD.absolute()))
+    yield board
+    backend.teardown_board(board)
+
+
+def test_constructor_board(board):
+    prot = board.fs.protocol
+
+    fs_name = prot if isinstance(prot, str) else prot[0]
+
+    if fs_name == "file":
+        con_name = "local"
+    elif fs_name == "rsc":
+        con_name = "rsconnect"
+        pytest.xfail()
+    else:
+        con_name = fs_name
+
+    board = getattr(c, f"board_{con_name}")(board.board)
+
+    # check cache
+    # check data
 
 
 # Board particulars ===========================================================