Merge pull request #22 from dwhswenson/command_tests

Command tests

Author: dwhswenson

paths_cli/commands/pathsampling.py

@@ -15,15 +15,15 @@
 @SCHEME.clicked(required=False)
 @INIT_CONDS.clicked(required=False)
 @N_STEPS_MC
-def path_sampling(input_file, output_file, scheme, init_conds, nsteps):
+def pathsampling(input_file, output_file, scheme, init_conds, nsteps):
     """General path sampling, using setup in INPUT_FILE"""
     storage = INPUT_FILE.get(input_file)
-    path_sampling_main(output_storage=OUTPUT_FILE.get(output_file),
-                       scheme=SCHEME.get(storage, scheme),
-                       init_conds=INIT_CONDS.get(storage, init_conds),
-                       n_steps=nsteps)
+    pathsampling_main(output_storage=OUTPUT_FILE.get(output_file),
+                      scheme=SCHEME.get(storage, scheme),
+                      init_conds=INIT_CONDS.get(storage, init_conds),
+                      n_steps=nsteps)
 
-def path_sampling_main(output_storage, scheme, init_conds, n_steps):
+def pathsampling_main(output_storage, scheme, init_conds, n_steps):
     import openpathsampling as paths
     init_conds = scheme.initial_conditions_from_trajectories(init_conds)
     simulation = paths.PathSampling(
@@ -37,6 +37,6 @@ def path_sampling_main(output_storage, scheme, init_conds, n_steps):
     return simulation.sample_set, simulation
 
 
-CLI = path_sampling
+CLI = pathsampling
 SECTION = "Simulation"
 REQUIRES_OPS = (1, 0)

paths_cli/tests/commands/test_contents.py

@@ -0,0 +1,42 @@
+import os
+import tempfile
+import pytest
+from unittest.mock import patch
+from click.testing import CliRunner
+
+import openpathsampling as paths
+
+from paths_cli.commands.contents import *
+
+def test_contents(tps_fixture):
+    # we just do a full integration test of this one
+    scheme, network, engine, init_conds = tps_fixture
+    runner = CliRunner()
+    with runner.isolated_filesystem():
+        storage = paths.Storage("setup.nc", 'w')
+        for obj in tps_fixture:
+            storage.save(obj)
+        storage.tags['initial_conditions'] = init_conds
+
+        results = runner.invoke(contents, ['setup.nc'])
+        cwd = os.getcwd()
+        expected = [
+            f"Storage @ '{cwd}/setup.nc'",
+            "CVs: 1 item", "* x",
+            "Volumes: 8 items", "* A", "* B", "* plus 6 unnamed items",
+            "Engines: 2 items", "* flat", "* plus 1 unnamed item",
+            "Networks: 1 item", "* 1 unnamed item",
+            "Move Schemes: 1 item", "* 1 unnamed item",
+            "Simulations: 0 items",
+            "Tags: 1 item", "* initial_conditions",
+            "", "Data Objects:",
+            "Steps: 0 unnamed items",
+            "Move Changes: 0 unnamed items",
+            "SampleSets: 1 unnamed item",
+            "Trajectories: 1 unnamed item",
+            f"Snapshots: {2*len(init_conds[0])} unnamed items", ""
+        ]
+        assert results.exit_code == 0
+        assert results.output.split('\n') == expected
+        for truth, beauty in zip(expected, results.output.split('\n')):
+            assert truth == beauty

paths_cli/tests/commands/test_equilibrate.py

@@ -0,0 +1,53 @@
+import os
+import pytest
+from unittest.mock import patch
+import tempfile
+from click.testing import CliRunner
+
+from paths_cli.commands.equilibrate import *
+
+import openpathsampling as paths
+
+def print_test(output_storage, scheme, init_conds, multiplier, extra_steps):
+    print(isinstance(output_storage, paths.Storage))
+    print(scheme.__uuid__)
+    print(init_conds.__uuid__)
+    print(multiplier, extra_steps)
+
+
+@patch('paths_cli.commands.equilibrate.equilibrate_main', print_test)
+def test_equilibrate(tps_fixture):
+    # integration test (click and parameters)
+    scheme, network, engine, init_conds = tps_fixture
+    runner = CliRunner()
+    with runner.isolated_filesystem():
+        storage = paths.Storage("setup.nc", 'w')
+        for obj in tps_fixture:
+            storage.save(obj)
+        storage.tags['initial_conditions'] = init_conds
+
+        results = runner.invoke(
+            equilibrate,
+            ["setup.nc", "-o", "foo.nc"]
+        )
+        out_str = "True\n{schemeid}\n{condsid}\n1 0\n"
+        expected_output = out_str.format(schemeid=scheme.__uuid__,
+                                         condsid=init_conds.__uuid__)
+        assert results.exit_code == 0
+        assert results.output == expected_output
+
+def test_equilibrate_main(tps_fixture):
+    # smoke test
+    tempdir = tempfile.mkdtemp()
+    store_name = os.path.join(tempdir, "equil.nc")
+    try:
+        storage = paths.Storage(store_name, mode='w')
+        scheme, network, engine, init_conds = tps_fixture
+        equilibrated, sim = equilibrate_main(storage, scheme, init_conds,
+                                             multiplier=1, extra_steps=1)
+        assert isinstance(equilibrated, paths.SampleSet)
+        assert isinstance(sim, paths.PathSampling)
+    finally:
+        if os.path.exists(store_name):
+            os.remove(store_name)
+        os.rmdir(tempdir)

paths_cli/tests/commands/test_pathsampling.py

@@ -0,0 +1,45 @@
+import os
+import pytest
+from unittest.mock import patch
+import tempfile
+from click.testing import CliRunner
+
+import openpathsampling as paths
+
+from paths_cli.commands.pathsampling import *
+
+def print_test(output_storage, scheme, init_conds, n_steps):
+    print(isinstance(output_storage, paths.Storage))
+    print(scheme.__uuid__)
+    print(init_conds.__uuid__)
+    print(n_steps)
+
+@patch('paths_cli.commands.pathsampling.pathsampling_main', print_test)
+def test_pathsampling(tps_fixture):
+    scheme, _, _, init_conds = tps_fixture
+    runner = CliRunner()
+    with runner.isolated_filesystem():
+        storage = paths.Storage("setup.nc", 'w')
+        for obj in tps_fixture:
+            storage.save(obj)
+        storage.tags['initial_conditions'] = init_conds
+
+        results = runner.invoke(pathsampling, ['setup.nc', '-o', 'foo.nc',
+                                               '-n', '1000'])
+        assert results.exit_code == 0
+        expected_output = (f"True\n{scheme.__uuid__}\n{init_conds.__uuid__}"
+                           "\n1000\n")
+        assert results.output == expected_output
+
+
+def test_pathsampling_main(tps_fixture):
+    scheme, _, _, init_conds = tps_fixture
+    with CliRunner().isolated_filesystem():
+        storage = paths.Storage("tps.nc", mode='w')
+        final, sim = pathsampling_main(storage, scheme, init_conds, 10)
+        assert isinstance(final, paths.SampleSet)
+        assert isinstance(sim, paths.PathSampling)
+        assert len(storage.steps) == 11
+        assert len(storage.schemes) == 1
+
+

paths_cli/tests/commands/test_visit_all.py

@@ -17,15 +17,6 @@ def print_test(output_storage, states, engine, initial_frame):
     print(initial_frame.__uuid__)
 
 
-@pytest.fixture()
-def tps_fixture(flat_engine, tps_network_and_traj):
-    network, traj = tps_network_and_traj
-    scheme = paths.OneWayShootingMoveScheme(network=network,
-                                            selector=paths.UniformSelector(),
-                                            engine=flat_engine)
-    init_conds = scheme.initial_conditions_from_trajectories(traj)
-    return (scheme, network, flat_engine, init_conds)
-
 @pytest.fixture()
 def visit_all_fixture(tps_fixture):
     scheme, network, engine, init_conds = tps_fixture

paths_cli/tests/conftest.py

@@ -23,3 +23,12 @@ def tps_network_and_traj():
     init_traj = make_1d_traj([-0.1, 0.1, 0.3, 0.5, 0.7, 0.9, 1.1],
                              [2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0])
     return (network, init_traj)
+
+@pytest.fixture
+def tps_fixture(flat_engine, tps_network_and_traj):
+    network, traj = tps_network_and_traj
+    scheme = paths.OneWayShootingMoveScheme(network=network,
+                                            selector=paths.UniformSelector(),
+                                            engine=flat_engine)
+    init_conds = scheme.initial_conditions_from_trajectories(traj)
+    return (scheme, network, flat_engine, init_conds)