Add tests for ReadOnlyRun

.github/scripts/run-e2e-tests.sh

@@ -21,7 +21,9 @@ EXIT_CODE=-1
 PROJECT="pye2e-fetcher-$(date +%Y-%m-%d_%H-%M-%S)-$RANDOM"
 
 cleanup() {
+  # Don't fail tests if cleanup fails
   set +e
+
   echo "Cleaning up..."
 
   echo "Deleting project $NEPTUNE_WORKSPACE/$PROJECT"
@@ -32,7 +34,7 @@ cleanup() {
 }
 
 # Make sure cleanup is called at exit
-trap cleanup SIGINT SIGTERM EXIT
+trap cleanup SIGINT SIGTERM EXIT ERR
 
 run_tests() {
   export NEPTUNE_PROJECT="$NEPTUNE_WORKSPACE/$PROJECT"

.github/workflows/tests-e2e.yml

@@ -28,7 +28,9 @@ jobs:
       - name: Run tests
         env:
           NEPTUNE_WORKSPACE: ${{ secrets.E2E_WORKSPACE }}
+          NEPTUNE_E2E_PROJECT: ${{ secrets.E2E_PROJECT }}
           NEPTUNE_API_TOKEN: ${{ secrets.E2E_API_TOKEN }}
+          NEPTUNE_E2E_CUSTOM_RUN_ID: ${{ vars.E2E_CUSTOM_RUN_ID }}
         run: bash .github/scripts/run-e2e-tests.sh
 
       - name: Upload test reports

tests/README.md

@@ -1,10 +1,32 @@
-# Important note
+# Important notes
 
-All tests in the `e2e` directory are end-to-end tests that assume a
+## Prepopulated test data
+
+A subset of the tests in the `e2e` directory are end-to-end tests that assume a
 specific data layout in a project.
 
 This data is populated using the `populate_projects.py` script, which
 has comments explaining the layout.
 
 The script *must* be run only **once** on a fresh project to set up the
 necessary data for the tests to run successfully.
+
+The tests that rely on this data are:
+
+* `test_dataframe_values.py`
+* `test_run_filtering.py`
+* `test_run_list.py`
+
+## Environment variables
+
+* `NEPTUNE_API_TOKEN` - API token to use
+* `NEPTUNE_E2E_FIXED_PROJECT` - project name to use for tests that require a project
+  with fixed data populated by `populate_projects.py` and **do not** populate the
+  project during execution. The test runner script creates a temporary project for
+  that purpose. If not set, `NEPTUNE_PROJECT` is used.
+* `NEPTUNE_E2E_PROJECT` - project name to use for tests that create Runs during
+  execution. This can be an existing project in which it's fine to create multiple
+  Runs with different data. If not set, `NEPTUNE_PROJECT` is used.
+* `NEPTUNE_E2E_CUSTOM_RUN_ID` (optional) - if set, it should be `sys/custom_run_id`
+  of an existing Run. This avoids creating a new Run for tests that log data,
+  if this is for some reason required.

tests/e2e/conftest.py

@@ -1,6 +1,17 @@
+import os
+import uuid
+from datetime import (
+    datetime,
+    timezone,
+)
+
+from neptune_scale import Run
 from pytest import fixture
 
-from neptune_fetcher import ReadOnlyProject
+from neptune_fetcher import (
+    ReadOnlyProject,
+    ReadOnlyRun,
+)
 
 # NOTE
 # The fixtures below assume that we're testing on a "many metrics" project
@@ -10,12 +21,18 @@
 # around names and values of the metadata.
 
 
-@fixture(scope="session")
-def project():
-    # Assume project name and API token are set in the environment.
-    # As ReadOnlyProject is basically stateless, we can reuse the same
-    # instance across all tests.
-    return ReadOnlyProject()
+@fixture(scope="module")
+def project(request):
+    # Assume the project name and API token are set in the environment using the standard
+    # NEPTUNE_PROJECT and NEPTUNE_API_TOKEN variables.
+    #
+    # Since ReadOnlyProject is essentially stateless, we can reuse the same
+    # instance across all tests in a module.
+    #
+    # We also allow overriding the project name per module by setting the
+    # module-level `NEPTUNE_PROJECT` variable.
+    project_name = getattr(request.module, "NEPTUNE_PROJECT", None)
+    return ReadOnlyProject(project=project_name)
 
 
 @fixture
@@ -50,3 +67,57 @@ def id_to_name():
     d |= {f"id-exp-{num}": f"exp{num}" for num in range(1, 7)}
 
     return d
+
+
+class SyncRun(Run):
+    """A neptune_scale.Run instance that waits for processing to complete
+    after each logging method call. This is useful for e2e tests, where we
+    usually want to wait for the data to be available before fetching it."""
+
+    def log(self, *args, **kwargs):
+        result = super().log(*args, **kwargs)
+        self.wait_for_processing()
+        return result
+
+
+@fixture(scope="module")
+def run_init_kwargs(project):
+    """Arguments to initialize a neptune_scale.Run instance"""
+
+    # TODO: if a test fails the run could be left in an indefinite state
+    #       Maybe we should just have it scoped 'function' and require passing
+    #       an existing run id
+    kwargs = {"project": project.project_identifier}
+    run_id = os.getenv("NEPTUNE_E2E_CUSTOM_RUN_ID")
+    if run_id is None:
+        run_id = str(uuid.uuid4())
+        kwargs["experiment_name"] = "pye2e-fetcher"
+    else:
+        kwargs["resume"] = True
+
+    kwargs["run_id"] = kwargs["family"] = run_id
+
+    return kwargs
+
+
+@fixture(scope="module")
+def run(project, run_init_kwargs):
+    """Plain neptune_scale.Run instance. We're scoping it to "module", as it seems to be a
+    good compromise, mostly because of execution time."""
+
+    run = Run(**run_init_kwargs)
+    run.log_configs({"test_start_time": datetime.now(timezone.utc)})
+
+    return run
+
+
+@fixture(scope="module")
+def sync_run(project, run_init_kwargs):
+    """Blocking run for logging data"""
+    return SyncRun(**run_init_kwargs)
+
+
+@fixture
+def ro_run(run, project):
+    """ReadOnlyRun pointing to the same run as the neptune_scale.Run"""
+    return ReadOnlyRun(read_only_project=project, custom_id=run._run_id)

tests/e2e/test_dataframe_values.py

@@ -1,7 +1,10 @@
+import os
 import re
 
 import pytest
 
+NEPTUNE_PROJECT = os.getenv("NEPTUNE_E2E_FIXED_PROJECT")
+
 
 @pytest.mark.parametrize("limit", [1, 2, 6, 12, 1000])
 def test__all_runs_limit(project, all_run_ids, all_experiment_ids, sys_columns, limit):

tests/e2e/test_read_only_run.py

@@ -0,0 +1,156 @@
+import os
+import random
+import time
+import uuid
+from datetime import (
+    datetime,
+    timezone,
+)
+
+from neptune_fetcher import ReadOnlyRun
+
+NEPTUNE_PROJECT = os.getenv("NEPTUNE_E2E_PROJECT")
+
+
+def unique_path(prefix):
+    return f"{prefix}__{datetime.now(timezone.utc).isoformat('-', 'seconds')}__{str(uuid.uuid4())[-4:]}"
+
+
+def refresh(ro_run: ReadOnlyRun):
+    """Create a new ReadOnlyRun instance with the same project and custom_id,
+    which is basically a "refresh" operation"""
+    return ReadOnlyRun(read_only_project=ro_run.project, custom_id=ro_run["sys/custom_run_id"].fetch())
+
+
+def random_series(length=10, start_step=0):
+    """Return a 2-tuple of step and value lists, both of length `length`"""
+    assert length > 0
+    assert start_step >= 0
+
+    j = random.random()
+    # Round to 0 to avoid floating point errors
+    steps = [round((j + x) ** 2.0, 0) for x in range(start_step, length)]
+    values = [round((j + x) ** 3.0, 0) for x in range(len(steps))]
+
+    return steps, values
+
+
+def test_atoms(sync_run, ro_run):
+    """Set atoms to a value, make sure it's equal when fetched"""
+
+    now = time.time()
+    data = {
+        "int-value": int(now),
+        "float-value": now,
+        "str-value": f"hello-{now}",
+        "true-value": True,
+        "false-value": False,
+        # The backend rounds the milliseconds component, so we're fine with just 0 to be more predictable
+        "datetime-value": datetime.now(timezone.utc).replace(microsecond=0),
+    }
+
+    sync_run.log_configs(data)
+
+    for key, value in data.items():
+        assert ro_run[key].fetch() == value, f"Value for {key} does not match"
+
+    # Replace the data and make sure the update is reflected AFTER we purge the cache for those items
+    updated_data = {
+        "int-value": int(now + 1),
+        "float-value": now + 1,
+        "str-value": f"hello-{now + 1}",
+        "true-value": False,
+        "false-value": True,
+        "datetime-value": datetime.now(timezone.utc).replace(year=1999, microsecond=0),
+    }
+
+    sync_run.log_configs(updated_data)
+
+    # The data should stay the same, as we haven't purged the cache yet
+    for key, value in data.items():
+        assert ro_run[key].fetch() == value, f"The cached value for {key} does not match"
+
+    # Now purge the cache for the logged items, and fetch them again. They should now have the new values
+    for key in data.keys():
+        del ro_run[key]
+
+    for key, value in updated_data.items():
+        assert ro_run[key].fetch() == value, f"The updated value for {key} does not match"
+
+
+def test_series_no_prefetch(run, ro_run):
+    path = unique_path("test_series/series_no_prefetch")
+
+    steps, values = random_series()
+
+    for step, value in zip(steps, values):
+        run.log_metrics(data={path: value}, step=step)
+
+    run.wait_for_processing()
+
+    df = ro_run[path].fetch_values()
+    assert df["step"].tolist() == steps
+    assert df["value"].tolist() == values
+
+
+def test_single_series_with_prefetch(run, ro_run):
+    path = unique_path("test_series/series_with_prefetch")
+
+    steps, values = random_series()
+
+    for step, value in zip(steps, values):
+        run.log_metrics(data={path: value}, step=step)
+
+    run.wait_for_processing()
+
+    ro_run.prefetch_series_values([path], use_threads=True)
+    df = ro_run[path].fetch_values()
+
+    assert df["step"].tolist() == steps
+    assert df["value"].tolist() == values
+
+
+def test_multiple_series_with_prefetch(run, ro_run):
+    path_base = unique_path("test_series/many_series_with_prefetch")
+    data = {f"{path_base}-{i}": i for i in range(20)}
+
+    run.log_metrics(data, step=1)
+    run.wait_for_processing()
+
+    ro_run = refresh(ro_run)
+    paths = [p for p in ro_run.field_names if p.startswith(path_base)]
+    assert len(paths) == len(data), "Not all data was logged"
+
+    ro_run.prefetch_series_values(paths, use_threads=True)
+    for path in paths:
+        df = ro_run[path].fetch_values()
+        assert df["step"].tolist() == [1]
+        assert df["value"].tolist() == [data[path]]
+
+
+def test_series_fetch_and_append(run, ro_run):
+    """Fetch a series, then append, then fetch again -- the new data points should be there"""
+
+    path = unique_path("test_series/series_no_prefetch")
+
+    steps, values = random_series()
+
+    for step, value in zip(steps, values):
+        run.log_metrics(data={path: value}, step=step)
+
+    run.wait_for_processing()
+
+    df = ro_run[path].fetch_values()
+    assert df["step"].tolist() == steps
+    assert df["value"].tolist() == values
+
+    steps2, values2 = random_series(length=5, start_step=len(steps))
+
+    for step, value in zip(steps2, values2):
+        run.log_metrics(data={path: value}, step=step)
+
+    run.wait_for_processing()
+
+    df = ro_run[path].fetch_values()
+    assert df["step"].tolist() == steps + steps2
+    assert df["value"].tolist() == values + values2

tests/e2e/test_run_filtering.py

@@ -1,9 +1,13 @@
+import os
+
 import pytest
 
 #
 # Tests for filtering runs by various attributes
 #
 
+NEPTUNE_PROJECT = os.getenv("NEPTUNE_E2E_FIXED_PROJECT")
+
 
 def test__runs_no_filter(project, all_run_ids, all_experiment_ids, sys_columns):
     """Requesting all runs without any filter should return all runs and experiments"""

tests/e2e/test_misc.py → tests/e2e/test_run_list.py

@@ -1,3 +1,8 @@
+import os
+
+NEPTUNE_PROJECT = os.getenv("NEPTUNE_E2E_FIXED_PROJECT")
+
+
 def test__list_runs(project, all_run_ids, all_experiment_ids, sys_columns_set):
     result = list(project.list_runs())
     ids = all_run_ids + all_experiment_ids

tests/populate_projects.py

@@ -55,16 +55,15 @@ def populate_run(run, run_id, tags=None):
         data = {f"metrics/foo{x + 1}": value for x in range(MM_NUM_FIELD_KIND)}
         data |= {f"metrics/bar{x + 1}": value for x in range(MM_NUM_FIELD_KIND)}
         data |= {f"metrics/bar{x + 1}-unique-{run_id}": value for x in range(10)}
-
-        run.log_metrics(step, data=data)
+        run.log_metrics(step=step, data=data)
 
     # Last step will have a predetermined value
     step += 1
     data = {f"metrics/foo{x + 1}": x + 1 for x in range(MM_NUM_FIELD_KIND)}
     data |= {f"metrics/bar{x + 1}-unique-{run_id}": x + 1 for x in range(10)}
     data |= {f"metrics/bar{x + 1}": x + 1 for x in range(MM_NUM_FIELD_KIND)}
 
-    run.log_metrics(step, data=data)
+    run.log_metrics(step=step, data=data)
 
 
 def populate_many_metrics(project):