diff --git a/SAS/TMSS/test/test_utils.py b/SAS/TMSS/test/test_utils.py
index 48073af773ad2ccd46db69209e29d70d06697396..658e74dcffb1e1c7b0d364ec0c70d66d4336ebf2 100644
--- a/SAS/TMSS/test/test_utils.py
+++ b/SAS/TMSS/test/test_utils.py
@@ -270,7 +270,8 @@ class TMSSTestEnvironment:
def __init__(self, host: str='127.0.0.1', preferred_django_port: int=8000, public_host: str=None,
exchange: str=os.environ.get("TMSS_EXCHANGE", DEFAULT_BUSNAME), broker: str=os.environ.get("TMSS_BROKER", DEFAULT_BROKER),
populate_schemas:bool=False, populate_test_data:bool=False,
- start_pg_listener: bool=True):
+ start_ra_test_environment: bool=False, start_pg_listener: bool=False,
+ start_subtask_scheduler: bool=False, start_dynamic_scheduler: bool=False):
self._exchange = exchange
self._broker = broker
self._populate_schemas = populate_schemas
@@ -284,9 +285,19 @@ class TMSSTestEnvironment:
public_host=public_host)
self.client_credentials = TemporaryCredentials(user=self.ldap_server.dbcreds.user,
password=self.ldap_server.dbcreds.password)
+
+ self._start_ra_test_environment = start_ra_test_environment
+ self.ra_test_environment = None
+
self._start_pg_listener = start_pg_listener
self.pg_listener = None
+ self._start_subtask_scheduler = start_subtask_scheduler
+ self.subtask_scheduler = None
+
+ self._start_dynamic_scheduler = start_dynamic_scheduler
+ self.dynamic_scheduler = None
+
# Check for correct Django version, should be at least 3.0
if django.VERSION[0] < 3:
print("\nWARNING: YOU ARE USING DJANGO VERSION '%s', WHICH WILL NOT SUPPORT ALL FEATURES IN TMSS!\n" %
@@ -318,12 +329,26 @@ class TMSSTestEnvironment:
user.is_superuser = True
user.save()
+ if self._start_ra_test_environment:
+ self.ra_test_environment = RATestEnvironment(exchange=self._exchange, broker=self._broker)
+ self.ra_test_environment.start()
+
if self._start_pg_listener:
# start the TMSSPGListener, so the changes in the database are posted as EventMessages on the bus
from lofar.sas.tmss.services.tmsspglistener import TMSSPGListener
self.pg_listener = TMSSPGListener(exchange=self._exchange, broker=self._broker, dbcreds=self.database.dbcreds)
self.pg_listener.start()
+ if self._start_subtask_scheduler:
+ from lofar.sas.tmss.services.subtask_scheduling import create_service as create_subtask_scheduling_service
+ self.subtask_scheduler = create_subtask_scheduling_service(exchange=self._exchange, broker=self._broker)
+ self.subtask_scheduler.start_listening()
+
+ if self._start_dynamic_scheduler:
+ from lofar.sas.tmss.services.dynamic_scheduling import create_dynamic_scheduling_service
+ self.dynamic_scheduler = create_dynamic_scheduling_service(exchange=self._exchange, broker=self._broker)
+ self.dynamic_scheduler.start_listening()
+
if self._populate_schemas or self._populate_test_data:
self.populate_schemas()
@@ -336,6 +361,18 @@ class TMSSTestEnvironment:
self.pg_listener.stop()
self.pg_listener = None
+ if self.subtask_scheduler is not None:
+ self.subtask_scheduler.stop_listening()
+ self.subtask_scheduler = None
+
+ if self.dynamic_scheduler is not None:
+ self.dynamic_scheduler.stop_listening()
+ self.dynamic_scheduler = None
+
+ if self.ra_test_environment is not None:
+ self.ra_test_environment.stop()
+ self.ra_test_environment = None
+
self.django_server.stop()
self.ldap_server.stop()
self.database.destroy()
@@ -406,9 +443,14 @@ def main_test_environment():
group.add_option("-P", "--public_host", dest="public_host", type="string", default='127.0.0.1',
help="expose the TMSS Django REST API via this host. [default=%default]")
- group = OptionGroup(parser, 'Example/Test data')
+ group = OptionGroup(parser, 'Example/Test data, schemas and services',
+ description='Options to enable/create example/test data, schemas and services. ' \
+ 'Without these options you get a lean and mean TMSS test environment, but then you need to run the background services yourselves, and create test data yourself. ' \
+ 'For standalone commissioning/testing/playing around you need all these options.')
parser.add_option_group(group)
group.add_option('-d', '--data', dest='data', action='store_true', help='populate the test-database with test/example data')
+ group.add_option('-s', '--schemas', dest='schemas', action='store_true', help='populate the test-database with the TMSS JSON schemas')
+ group.add_option('-S', '--services', dest='services', action='store_true', help='start the TMSS background services.')
group = OptionGroup(parser, 'Messaging options')
parser.add_option_group(group)
@@ -419,40 +461,36 @@ def main_test_environment():
logging.basicConfig(format = '%(asctime)s %(levelname)s %(message)s', level = logging.INFO)
- with RATestEnvironment(exchange=options.exchange, broker=options.broker):
- with TMSSTestEnvironment(host=options.host, preferred_django_port=options.port, public_host=options.public_host,
- exchange=options.exchange, broker=options.broker,
- populate_schemas=True, populate_test_data=False) as tmss_test_env:
-
- from lofar.sas.tmss.services.dynamic_scheduling import create_dynamic_scheduling_service
- with create_dynamic_scheduling_service(options.exchange, options.broker):
- if options.data:
- tmss_test_env.populate_test_data()
-
- # print some nice info for the user to use the test servers...
- # use print instead of log for clean lines.
- for h in logging.root.handlers:
- h.flush()
- print()
- print()
- print("*****************************************************")
- print("Test-TMSS database, LDAP and Django up and running...")
- print("*****************************************************")
- print("DB Credentials ID: %s" % (tmss_test_env.database.dbcreds_id, ))
- print("LDAP Credentials ID: %s" % (tmss_test_env.django_server.ldap_dbcreds_id, ))
- print("TMSS Client Credentials ID: %s" % (tmss_test_env.client_credentials.dbcreds_id, ))
- print("Django URL: %s" % (tmss_test_env.django_server.url))
- print()
- print("Example cmdlines to run tmss or tmss_manage_django:")
- print("TMSS_DBCREDENTIALS=%s TMSS_LDAPCREDENTIALS=%s tmss" % (tmss_test_env.database.dbcreds_id, tmss_test_env.django_server.ldap_dbcreds_id))
- print("TMSS_DBCREDENTIALS=%s TMSS_LDAPCREDENTIALS=%s tmss_manage_django" % (tmss_test_env.database.dbcreds_id, tmss_test_env.django_server.ldap_dbcreds_id))
- print()
- print("Example cmdline to run tmss client call:")
- print("TMSS_CLIENT_DBCREDENTIALS=%s tmss_set_subtask_state <id> <state>" % (tmss_test_env.client_credentials.dbcreds_id, ))
- print()
- print("Press Ctrl-C to exit (and remove the test database and django server automatically)")
-
- waitForInterrupt()
+ with TMSSTestEnvironment(host=options.host, preferred_django_port=options.port, public_host=options.public_host,
+ exchange=options.exchange, broker=options.broker,
+ populate_schemas=options.schemas, populate_test_data=options.data,
+ start_ra_test_environment=options.services, start_pg_listener=options.services,
+ start_subtask_scheduler=options.services, start_dynamic_scheduler=options.services) as tmss_test_env:
+
+ # print some nice info for the user to use the test servers...
+ # use print instead of log for clean lines.
+ for h in logging.root.handlers:
+ h.flush()
+ print()
+ print()
+ print("*****************************************************")
+ print("Test-TMSS database, LDAP and Django up and running...")
+ print("*****************************************************")
+ print("DB Credentials ID: %s" % (tmss_test_env.database.dbcreds_id, ))
+ print("LDAP Credentials ID: %s" % (tmss_test_env.django_server.ldap_dbcreds_id, ))
+ print("TMSS Client Credentials ID: %s" % (tmss_test_env.client_credentials.dbcreds_id, ))
+ print("Django URL: %s" % (tmss_test_env.django_server.url))
+ print()
+ print("Example cmdlines to run tmss or tmss_manage_django:")
+ print("TMSS_DBCREDENTIALS=%s TMSS_LDAPCREDENTIALS=%s tmss" % (tmss_test_env.database.dbcreds_id, tmss_test_env.django_server.ldap_dbcreds_id))
+ print("TMSS_DBCREDENTIALS=%s TMSS_LDAPCREDENTIALS=%s tmss_manage_django" % (tmss_test_env.database.dbcreds_id, tmss_test_env.django_server.ldap_dbcreds_id))
+ print()
+ print("Example cmdline to run tmss client call:")
+ print("TMSS_CLIENT_DBCREDENTIALS=%s tmss_set_subtask_state <id> <state>" % (tmss_test_env.client_credentials.dbcreds_id, ))
+ print()
+ print("Press Ctrl-C to exit (and remove the test database and django server automatically)")
+
+ waitForInterrupt()
if __name__ == '__main__':