python_packaging/sample_restraint/tests/test_plugin.py

   1 # The myplugin module must be locatable by Python.
   2 # If you configured CMake in the build directory ``/path/to/repo/build`` then,
   3 # assuming you are in ``/path/to/repo``, run the tests with something like
   4 #     PYTHONPATH=./cmake-build-debug/src/pythonmodule mpiexec -n 2 python -m mpi4py -m pytest tests/
   5
   6 # This test is not currently run automatically in any way. Build the module, point your PYTHONPATH at it,
   7 # and run pytest in the tests directory.
   8
   9 import logging
  10 import os
  11
  12 try:
  13     import mpi4py.MPI as _MPI
  14 except (ImportError, ModuleNotFoundError):
  15     _MPI = None
  16
  17 import gmxapi as gmx
  18 from gmxapi.simulation.context import Context
  19 from gmxapi.simulation.workflow import WorkElement, from_tpr
  20 from gmxapi import version as gmx_version
  21 import pytest
  22
  23 # create console handler
  24 ch = logging.StreamHandler()
  25 ch.setLevel(logging.DEBUG)
  26 # create formatter and add it to the handler
  27 formatter = logging.Formatter('%(asctime)s:%(name)s:%(levelname)s: %(message)s')
  28 ch.setFormatter(formatter)
  29 # add the handlers to the logger
  30 logging.getLogger().addHandler(ch)
  31
  32 logger = logging.getLogger()
  33
  34
  35 def test_import():
  36     # Suppress inspection warning outside of testing context.
  37     # noinspection PyUnresolvedReferences
  38     import myplugin
  39     assert myplugin
  40
  41
  42 @pytest.mark.usefixtures("cleandir")
  43 def test_binding_protocol(spc_water_box, mdrun_kwargs):
  44     """Test that gmxapi successfully attaches MD plugins."""
  45     import myplugin
  46
  47     if _MPI is not None:
  48         _size = _MPI.COMM_WORLD.Get_size()
  49         _rank = _MPI.COMM_WORLD.Get_rank()
  50     else:
  51         _size = 1
  52         _rank = 0
  53
  54     tpr_filename = spc_water_box
  55     logger.info("Testing plugin potential with input file {}".format(os.path.abspath(tpr_filename)))
  56
  57     assert gmx.version.api_is_at_least(0, 2, 1)
  58     md = from_tpr([tpr_filename] * _size, append_output=False, **mdrun_kwargs)
  59
  60     potential = WorkElement(namespace="myplugin",
  61                             operation="null_restraint",
  62                             params={'sites': [1, 4]})
  63     potential.name = "null restraint"
  64     md.add_dependency(potential)
  65
  66     context = Context(md)
  67
  68     with context as session:
  69         session.run()
  70
  71     # See also #3038, #3145, #4079
  72     assert isinstance(context.potentials, list)
  73     assert len(context.potentials) > 0
  74     for restraint in context.potentials:
  75         if isinstance(restraint, myplugin.NullRestraint):
  76             assert gmx.version.api_is_at_least(0, 2, 1)
  77             assert restraint.count() > 1
  78
  79
  80 @pytest.mark.usefixtures("cleandir")
  81 def test_ensemble_potential_nompi(spc_water_box, mdrun_kwargs):
  82     """Test ensemble potential without an ensemble.
  83     """
  84     tpr_filename = spc_water_box
  85     logger.info("Testing plugin potential with input file {}".format(os.path.abspath(tpr_filename)))
  86
  87     assert gmx.version.api_is_at_least(0, 0, 5)
  88     md = from_tpr([tpr_filename], append_output=False, **mdrun_kwargs)
  89
  90     # Create a WorkElement for the potential
  91     params = {'sites': [1, 4],
  92               'nbins': 10,
  93               'binWidth': 0.1,
  94               'min_dist': 0.,
  95               'max_dist': 10.,
  96               'experimental': [1.] * 10,
  97               'nsamples': 1,
  98               'sample_period': 0.001,
  99               'nwindows': 4,
 100               'k': 10000.,
 101               'sigma': 1.}
 102     potential = WorkElement(namespace="myplugin",
 103                             operation="ensemble_restraint",
 104                             params=params)
 105     # Note that we could flexibly capture accessor methods as workflow elements, too. Maybe we can
 106     # hide the extra Python bindings by letting myplugin.HarmonicRestraint automatically convert
 107     # to a WorkElement when add_dependency is called on it.
 108     potential.name = "ensemble_restraint"
 109     md.add_dependency(potential)
 110
 111     context = Context(md)
 112
 113     with context as session:
 114         session.run()
 115
 116
 117 @pytest.mark.withmpi_only
 118 @pytest.mark.usefixtures("cleandir")
 119 def test_ensemble_potential_withmpi(spc_water_box, mdrun_kwargs):
 120     tpr_filename = spc_water_box
 121
 122     logger.info("Testing plugin potential with input file {}".format(os.path.abspath(tpr_filename)))
 123
 124     assert gmx_version.api_is_at_least(0, 0, 5)
 125     md = from_tpr([tpr_filename, tpr_filename], append_output=False, **mdrun_kwargs)
 126
 127     # Create a WorkElement for the potential
 128     params = {'sites': [1, 4],
 129               'nbins': 10,
 130               'binWidth': 0.1,
 131               'min_dist': 0.,
 132               'max_dist': 10.,
 133               'experimental': [0.5] * 10,
 134               'nsamples': 1,
 135               'sample_period': 0.001,
 136               'nwindows': 4,
 137               'k': 10000.,
 138               'sigma': 1.}
 139
 140     potential = WorkElement(namespace="myplugin",
 141                             operation="ensemble_restraint",
 142                             params=params)
 143     # Note that we could flexibly capture accessor methods as workflow elements, too. Maybe we can
 144     # hide the extra Python bindings by letting myplugin.HarmonicRestraint automatically convert
 145     # to a WorkElement when add_dependency is called on it.
 146     potential.name = "ensemble_restraint"
 147     md.add_dependency(potential)
 148
 149     context = Context(md)
 150     with context as session:
 151         session.run()