Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions src/openfe/tests/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -161,17 +161,17 @@ def other_mapping():


@pytest.fixture()
def lomap_basic_test_files_dir(tmpdir_factory):
def lomap_basic_test_files_dir(tmp_path_factory):
# for lomap, which wants the files in a directory
lomap_files = tmpdir_factory.mktemp("lomap_files")
lomap_files = tmp_path_factory.mktemp("lomap_files")
lomap_basic = "openfe.tests.data.lomap_basic"

for f in resources.contents(lomap_basic):
if not f.endswith("mol2"):
continue
stuff = resources.read_binary(lomap_basic, f)

with open(str(lomap_files.join(f)), "wb") as fout:
with open(lomap_files / f, "wb") as fout:
fout.write(stuff)

yield str(lomap_files)
Expand Down
9 changes: 5 additions & 4 deletions src/openfe/tests/protocols/openmm_abfe/test_abfe_energies.py
Original file line number Diff line number Diff line change
Expand Up @@ -117,7 +117,7 @@ def t4_reference_system(self):
return system

@pytest.fixture()
def t4_validation_data(self, benzene_modifications, T4_protein_component, tmpdir):
def t4_validation_data(self, benzene_modifications, T4_protein_component, tmp_path):
s = openmm_afe.AbsoluteBindingProtocol.default_settings()
s.protocol_repeats = 1
s.engine_settings.compute_platform = "cpu"
Expand Down Expand Up @@ -149,9 +149,10 @@ def t4_validation_data(self, benzene_modifications, T4_protein_component, tmpdir

complex_units = [u for u in dag.protocol_units if isinstance(u, ABFEComplexSetupUnit)]

with tmpdir.as_cwd():
results = complex_units[0].run(dry=True)
return results
results = complex_units[0].run(
dry=True, scratch_basepath=tmp_path, shared_basepath=tmp_path
)
return results

@staticmethod
def get_energy_components(
Expand Down
239 changes: 127 additions & 112 deletions src/openfe/tests/protocols/openmm_abfe/test_abfe_protocol.py
Original file line number Diff line number Diff line change
Expand Up @@ -386,100 +386,112 @@ def _test_energies(reference_system, alchemical_system, alchemical_regions, posi
positions=positions,
)

def test_complex_dry_run(self, complex_setup_units, complex_sim_units, settings, tmpdir):
with tmpdir.as_cwd():
setup_results = complex_setup_units[0].run(dry=True, verbose=True)
sim_results = complex_sim_units[0].run(
system=setup_results["alchem_system"],
positions=setup_results["debug_positions"],
selection_indices=setup_results["selection_indices"],
box_vectors=setup_results["box_vectors"],
alchemical_restraints=True,
dry=True,
)

# Check the sampler
self._verify_sampler(sim_results["sampler"], complexed=True, settings=settings)

# Check the alchemical system
self._assert_expected_alchemical_forces(
setup_results["alchem_system"], complexed=True, settings=settings
)
self._test_dodecahedron_vectors(setup_results["alchem_system"])

# Check the alchemical indices
expected_indices = [i + self.num_complex_atoms for i in range(self.num_solvent_atoms)]
assert expected_indices == setup_results["alchem_indices"]

# Check the non-alchemical system
self._assert_expected_nonalchemical_forces(setup_results["standard_system"], settings)
self._test_dodecahedron_vectors(setup_results["standard_system"])
# Check the box vectors haven't changed (they shouldn't have because we didn't do MD)
assert_allclose(
from_openmm(setup_results["alchem_system"].getDefaultPeriodicBoxVectors()),
from_openmm(setup_results["standard_system"].getDefaultPeriodicBoxVectors()),
)

# Check the PDB
pdb = mdt.load_pdb(setup_results["pdb_structure"])
assert pdb.n_atoms == self.num_all_not_water

# Check energies
alchem_region = AlchemicalRegion(alchemical_atoms=setup_results["alchem_indices"])
self._test_energies(
reference_system=setup_results["standard_system"],
alchemical_system=setup_results["alchem_system"],
alchemical_regions=alchem_region,
positions=setup_results["debug_positions"],
)

def test_solvent_dry_run(self, solvent_setup_units, solvent_sim_units, settings, tmpdir):
with tmpdir.as_cwd():
setup_results = solvent_setup_units[0].run(dry=True, verbose=True)
sim_results = solvent_sim_units[0].run(
system=setup_results["alchem_system"],
positions=setup_results["debug_positions"],
selection_indices=setup_results["selection_indices"],
box_vectors=setup_results["box_vectors"],
alchemical_restraints=False,
dry=True,
)

# Check the sampler
self._verify_sampler(sim_results["sampler"], complexed=False, settings=settings)

# Check the alchemical system
self._assert_expected_alchemical_forces(
setup_results["alchem_system"], complexed=False, settings=settings
)
self._test_cubic_vectors(setup_results["alchem_system"])

# Check the alchemical indices
expected_indices = [i for i in range(self.num_solvent_atoms)]
assert expected_indices == setup_results["alchem_indices"]

# Check the non-alchemical system
self._assert_expected_nonalchemical_forces(setup_results["standard_system"], settings)
self._test_cubic_vectors(setup_results["standard_system"])
# Check the box vectors haven't changed (they shouldn't have because we didn't do MD)
assert_allclose(
from_openmm(setup_results["alchem_system"].getDefaultPeriodicBoxVectors()),
from_openmm(setup_results["standard_system"].getDefaultPeriodicBoxVectors()),
)

# Check the PDB
pdb = mdt.load_pdb(setup_results["pdb_structure"])
assert pdb.n_atoms == self.num_solvent_atoms

# Check energies
alchem_region = AlchemicalRegion(alchemical_atoms=setup_results["alchem_indices"])

self._test_energies(
reference_system=setup_results["standard_system"],
alchemical_system=setup_results["alchem_system"],
alchemical_regions=alchem_region,
positions=setup_results["debug_positions"],
)
def test_complex_dry_run(self, complex_setup_units, complex_sim_units, settings, tmp_path):
setup_results = complex_setup_units[0].run(
dry=True,
verbose=True,
scratch_basepath=tmp_path,
shared_basepath=tmp_path,
)
sim_results = complex_sim_units[0].run(
system=setup_results["alchem_system"],
positions=setup_results["debug_positions"],
selection_indices=setup_results["selection_indices"],
box_vectors=setup_results["box_vectors"],
alchemical_restraints=True,
dry=True,
scratch_basepath=tmp_path,
shared_basepath=tmp_path,
)

# Check the sampler
self._verify_sampler(sim_results["sampler"], complexed=True, settings=settings)

# Check the alchemical system
self._assert_expected_alchemical_forces(
setup_results["alchem_system"], complexed=True, settings=settings
)
self._test_dodecahedron_vectors(setup_results["alchem_system"])

# Check the alchemical indices
expected_indices = [i + self.num_complex_atoms for i in range(self.num_solvent_atoms)]
assert expected_indices == setup_results["alchem_indices"]

# Check the non-alchemical system
self._assert_expected_nonalchemical_forces(setup_results["standard_system"], settings)
self._test_dodecahedron_vectors(setup_results["standard_system"])
# Check the box vectors haven't changed (they shouldn't have because we didn't do MD)
assert_allclose(
from_openmm(setup_results["alchem_system"].getDefaultPeriodicBoxVectors()),
from_openmm(setup_results["standard_system"].getDefaultPeriodicBoxVectors()),
)

# Check the PDB
pdb = mdt.load_pdb(setup_results["pdb_structure"])
assert pdb.n_atoms == self.num_all_not_water

# Check energies
alchem_region = AlchemicalRegion(alchemical_atoms=setup_results["alchem_indices"])
self._test_energies(
reference_system=setup_results["standard_system"],
alchemical_system=setup_results["alchem_system"],
alchemical_regions=alchem_region,
positions=setup_results["debug_positions"],
)

def test_solvent_dry_run(self, solvent_setup_units, solvent_sim_units, settings, tmp_path):
setup_results = solvent_setup_units[0].run(
dry=True,
verbose=True,
scratch_basepath=tmp_path,
shared_basepath=tmp_path,
)
sim_results = solvent_sim_units[0].run(
system=setup_results["alchem_system"],
positions=setup_results["debug_positions"],
selection_indices=setup_results["selection_indices"],
box_vectors=setup_results["box_vectors"],
alchemical_restraints=False,
dry=True,
scratch_basepath=tmp_path,
shared_basepath=tmp_path,
)

# Check the sampler
self._verify_sampler(sim_results["sampler"], complexed=False, settings=settings)

# Check the alchemical system
self._assert_expected_alchemical_forces(
setup_results["alchem_system"], complexed=False, settings=settings
)
self._test_cubic_vectors(setup_results["alchem_system"])

# Check the alchemical indices
expected_indices = [i for i in range(self.num_solvent_atoms)]
assert expected_indices == setup_results["alchem_indices"]

# Check the non-alchemical system
self._assert_expected_nonalchemical_forces(setup_results["standard_system"], settings)
self._test_cubic_vectors(setup_results["standard_system"])
# Check the box vectors haven't changed (they shouldn't have because we didn't do MD)
assert_allclose(
from_openmm(setup_results["alchem_system"].getDefaultPeriodicBoxVectors()),
from_openmm(setup_results["standard_system"].getDefaultPeriodicBoxVectors()),
)

# Check the PDB
pdb = mdt.load_pdb(setup_results["pdb_structure"])
assert pdb.n_atoms == self.num_solvent_atoms

# Check energies
alchem_region = AlchemicalRegion(alchemical_atoms=setup_results["alchem_indices"])

self._test_energies(
reference_system=setup_results["standard_system"],
alchemical_system=setup_results["alchem_system"],
alchemical_regions=alchem_region,
positions=setup_results["debug_positions"],
)


@pytest.mark.slow
Expand Down Expand Up @@ -511,7 +523,7 @@ def settings(self):
return s


def test_user_charges(benzene_modifications, T4_protein_component, tmpdir):
def test_user_charges(benzene_modifications, T4_protein_component, tmp_path):
s = openmm_afe.AbsoluteBindingProtocol.default_settings()
s.protocol_repeats = 1
s.engine_settings.compute_platform = "cpu"
Expand Down Expand Up @@ -558,24 +570,27 @@ def assign_fictitious_charges(offmol):

complex_setup_units = _get_units(dag.protocol_units, UNIT_TYPES["complex"]["setup"])

with tmpdir.as_cwd():
results = complex_setup_units[0].run(dry=True)
results = complex_setup_units[0].run(
dry=True,
scratch_basepath=tmp_path,
shared_basepath=tmp_path,
)

system_nbf = [
f for f in results["standard_system"].getForces() if isinstance(f, NonbondedForce)
][0]
alchem_system_nbf = [
f
for f in results["alchem_system"].getForces()
if isinstance(f, NonbondedForce)
][0] # fmt: skip
system_nbf = [
f for f in results["standard_system"].getForces() if isinstance(f, NonbondedForce)
][0]
alchem_system_nbf = [
f
for f in results["alchem_system"].getForces()
if isinstance(f, NonbondedForce)
][0] # fmt: skip

for i in range(12):
# add 2613 to account for the protein
index = i + 2613
for i in range(12):
# add 2613 to account for the protein
index = i + 2613

c, s, e = system_nbf.getParticleParameters(index)
assert pytest.approx(prop_chgs[i]) == c.value_in_unit(ommunit.elementary_charge)
c, s, e = system_nbf.getParticleParameters(index)
assert pytest.approx(prop_chgs[i]) == c.value_in_unit(ommunit.elementary_charge)

offsets = alchem_system_nbf.getParticleParameterOffset(i)
assert pytest.approx(prop_chgs[i]) == offsets[2]
offsets = alchem_system_nbf.getParticleParameterOffset(i)
assert pytest.approx(prop_chgs[i]) == offsets[2]
Original file line number Diff line number Diff line change
Expand Up @@ -83,12 +83,12 @@ def patcher():
yield


def test_gather(benzene_complex_dag, patcher, tmpdir):
def test_gather(benzene_complex_dag, patcher, tmp_path):
# check that .gather behaves as expected
dagres = gufe.protocols.execute_DAG(
benzene_complex_dag,
shared_basedir=tmpdir,
scratch_basedir=tmpdir,
shared_basedir=tmp_path,
scratch_basedir=tmp_path,
keep_shared=True,
)

Expand All @@ -101,7 +101,7 @@ def test_gather(benzene_complex_dag, patcher, tmpdir):
assert isinstance(res, openmm_afe.AbsoluteBindingProtocolResult)


def test_unit_tagging(benzene_complex_dag, patcher, tmpdir):
def test_unit_tagging(benzene_complex_dag, patcher, tmp_path):
# test that executing the units includes correct gen and repeat info

dag_units = benzene_complex_dag.protocol_units
Expand All @@ -117,19 +117,19 @@ def test_unit_tagging(benzene_complex_dag, patcher, tmpdir):

for u in setup_units:
rid = u.inputs["repeat_id"]
setup_results[rid] = u.execute(context=gufe.Context(tmpdir, tmpdir))
setup_results[rid] = u.execute(context=gufe.Context(tmp_path, tmp_path))

for u in sim_units:
rid = u.inputs["repeat_id"]
sim_results[rid] = u.execute(
context=gufe.Context(tmpdir, tmpdir),
context=gufe.Context(tmp_path, tmp_path),
setup_results=setup_results[rid],
)

for u in a_units:
rid = u.inputs["repeat_id"]
analysis_results[rid] = u.execute(
context=gufe.Context(tmpdir, tmpdir),
context=gufe.Context(tmp_path, tmp_path),
setup_results=setup_results[rid],
simulation_results=sim_results[rid],
)
Expand Down
8 changes: 4 additions & 4 deletions src/openfe/tests/protocols/openmm_abfe/test_abfe_slow.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ def test_openmm_run_engine(
eg5_protein,
eg5_ligands,
eg5_cofactor,
tmpdir,
tmp_path,
):
if platform not in available_platforms:
pytest.skip(f"OpenMM Platform: {platform} not available")
Expand Down Expand Up @@ -92,7 +92,7 @@ def test_openmm_run_engine(
mapping=None,
)

cwd = pathlib.Path(str(tmpdir))
cwd = pathlib.Path(tmp_path)
r = openfe.execute_DAG(dag, shared_basedir=cwd, scratch_basedir=cwd, keep_shared=True)

assert r.ok()
Expand All @@ -104,7 +104,7 @@ def test_openmm_run_engine(
# get the path to the simulation unit shared dict
for pur in purs:
if "Simulation" in pur.name:
sim_shared = tmpdir / f"shared_{pur.source_key}_attempt_0"
sim_shared = tmp_path / f"shared_{pur.source_key}_attempt_0"
assert sim_shared.exists()
assert pathlib.Path(sim_shared).is_dir()

Expand All @@ -113,7 +113,7 @@ def test_openmm_run_engine(
if "Analysis" not in pur.name:
continue

unit_shared = tmpdir / f"shared_{pur.source_key}_attempt_0"
unit_shared = tmp_path / f"shared_{pur.source_key}_attempt_0"
assert unit_shared.exists()
assert pathlib.Path(unit_shared).is_dir()

Expand Down
Loading
Loading