Tickets/dm41263: Parameterized OCPS index based on test environment

.gitignore

@@ -30,3 +30,6 @@ ospl*log
 .flake8
 .isort.cfg
 .mypy.ini
+.clang-format
+.ruff.toml
+towncrier.toml

doc/version-history.rst

@@ -10,6 +10,10 @@ Version History
 .. No new work should be required in order to complete this section.
 .. Below is an example of a version history format.
 
+v0.17.0
+-------
+* Parameterized the OCSP 2||3 index, determined by test environment.
+
 v0.16.0
 -------
 * Added Watcher to the ObsSys State transition tests.

python/lsst/ts/IntegrationTests/__init__.py

@@ -64,4 +64,5 @@
 from .obssys_standby_disabled import *
 from .script_queue_controller import *
 from .testutils import *
+from .utils import *
 from .yaml_test_strings import *

python/lsst/ts/IntegrationTests/configs/enabled_offline_state_transition_configs.py

@@ -46,7 +46,7 @@
     - [Scheduler:1, OFFLINE]
     - [Scheduler:2, OFFLINE]
     - [OCPS:1, OFFLINE]
-    - [OCPS:2, OFFLINE]
+    - [replace_me, OFFLINE]
     """
 )
 

python/lsst/ts/IntegrationTests/configs/obssys_state_transition_configs.py

@@ -34,7 +34,7 @@
     - [Scheduler:1, DISABLED]
     - [Scheduler:2, DISABLED]
     - [OCPS:1, DISABLED]
-    - [OCPS:2, DISABLED]
+    - [replace_me, DISABLED]
     - [Watcher, DISABLED]
     """
 )
@@ -50,7 +50,7 @@
     - [Scheduler:1, ENABLED]
     - [Scheduler:2, ENABLED]
     - [OCPS:1, ENABLED]
-    - [OCPS:2, ENABLED]
+    - [replace_me, ENABLED]
     - [Watcher, ENABLED]
     """
 )

python/lsst/ts/IntegrationTests/enabled_offline.py

@@ -22,8 +22,10 @@
 
 __all__ = ["EnabledOffline", "run_enabled_offline"]
 
+import argparse
 import asyncio
 
+import yaml
 from lsst.ts.IntegrationTests import BaseScript
 
 from .configs.config_registry import registry
@@ -63,12 +65,64 @@ class EnabledOffline(BaseScript):
         ("set_summary_state.py", BaseScript.is_standard),
     ]
 
-    def __init__(self) -> None:
+    def __init__(self, test_env: str) -> None:
         super().__init__()
+        # Set the OCPS index based on test environment
+        self.test_env = test_env
+        self.env_configs = yaml.safe_load(self.configs[1])
+        if test_env.lower() == "bts":
+            # Running on BTS with OCPS:3
+            self.ocps = "OCPS:3"
+        else:
+            # Running on TTS or Summit with OCPS:2
+            self.ocps = "OCPS:2"
+        self.env_configs["data"][3][0] = self.ocps
+        # Update the self.configs tuple with the updated
+        # registry["sched_ocps_enabled_offline"] configuration.
+        # Do this by converting the tuple to a list, replacing the
+        # updated entry and converting it back to a tuple.
+        temp_list = list(self.configs)
+        temp_list[1] = yaml.safe_dump(
+            self.env_configs, explicit_start=True, canonical=True
+        )
+        self.configs = tuple(temp_list)
 
 
 def run_enabled_offline() -> None:
-    script_class = EnabledOffline()
-    num_scripts = len(script_class.scripts)
-    print(f"\nEnabled to Offline; running {num_scripts} scripts")
-    asyncio.run(script_class.run())
+    # Define the script arguments.
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "test_env",
+        nargs="?",
+        type=str.lower,
+        choices=["bts", "tts", "summit"],
+        help="Specify on which environment the tests are running (case insensitive).",
+    )
+    args = parser.parse_args()
+    # Print the help if the environment is not defined.
+    if not (args.test_env):
+        parser.print_help()
+        exit()
+    main(args)
+
+
+def main(opts: argparse.Namespace) -> None:
+    # Ensure the invocation is correct.
+    # If not, raise KeyError.
+    # If it is correct, execute the state transition.
+    try:
+        script_class = EnabledOffline(
+            test_env=opts.test_env,
+        )
+    except KeyError as ke:
+        print(repr(ke))
+    else:
+        num_scripts = len(script_class.scripts)
+        print(
+            f"\nEnabled to Offline; "
+            f"running {num_scripts} scripts "
+            f"on the '{opts.test_env}' environment. "
+            f"with this configuration: \n"
+            f"{script_class.configs}"
+        )
+        asyncio.run(script_class.run())

python/lsst/ts/IntegrationTests/obssys_disabled_enabled.py

@@ -24,9 +24,11 @@
 
 import asyncio
 
+import yaml
 from lsst.ts.IntegrationTests import BaseScript
 
 from .configs.config_registry import registry
+from .utils import get_test_env_arg
 
 
 class ObsSysDisabledEnabled(BaseScript):
@@ -42,12 +44,39 @@ class ObsSysDisabledEnabled(BaseScript):
         ("set_summary_state.py", BaseScript.is_standard),
     ]
 
-    def __init__(self) -> None:
+    def __init__(self, test_env: str) -> None:
         super().__init__()
+        # Set the OCPS index based on test environment
+        self.test_env = test_env
+        self.env_configs = yaml.safe_load(registry["obssys_disabled_enabled"])
+        if test_env.lower() == "bts":
+            # Running on BTS with OCPS:3
+            self.ocps = "OCPS:3"
+        else:
+            # Running on TTS or Summit with OCPS:2
+            self.ocps = "OCPS:2"
+        self.env_configs["data"][3][0] = self.ocps
+        self.configs = (yaml.safe_dump(self.env_configs),)
 
 
 def run_obssys_disabled_enabled() -> None:
-    script_class = ObsSysDisabledEnabled()
-    num_scripts = len(script_class.scripts)
-    print(f"\nObsSys Disabled to Enabled; running {num_scripts} scripts")
-    asyncio.run(script_class.run())
+    # Ensure the invocation is correct.
+    # If not, raise KeyError.
+    # If it is correct, execute the state transition.
+    args = get_test_env_arg()
+    try:
+        script_class = ObsSysDisabledEnabled(
+            test_env=args.test_env,
+        )
+    except KeyError as ke:
+        print(repr(ke))
+    else:
+        num_scripts = len(script_class.scripts)
+        print(
+            f"\nObsSys Disabled to Enabled; "
+            f"running {num_scripts} scripts "
+            f"on the '{args.test_env}' environment, "
+            f"with this configuration: \n"
+            f"{script_class.configs}"
+        )
+        asyncio.run(script_class.run())

python/lsst/ts/IntegrationTests/obssys_standby_disabled.py

@@ -24,9 +24,11 @@
 
 import asyncio
 
+import yaml
 from lsst.ts.IntegrationTests import BaseScript
 
 from .configs.config_registry import registry
+from .utils import get_test_env_arg
 
 
 class ObsSysStandbyDisabled(BaseScript):
@@ -42,12 +44,39 @@ class ObsSysStandbyDisabled(BaseScript):
         ("set_summary_state.py", BaseScript.is_standard),
     ]
 
-    def __init__(self) -> None:
+    def __init__(self, test_env: str) -> None:
         super().__init__()
+        # Set the OCPS index based on test environment
+        self.test_env = test_env
+        self.env_configs = yaml.safe_load(registry["obssys_standby_disabled"])
+        if test_env.lower() == "bts":
+            # Running on BTS with OCPS:3
+            self.ocps = "OCPS:3"
+        else:
+            # Running on TTS or Summit with OCPS:2
+            self.ocps = "OCPS:2"
+        self.env_configs["data"][3][0] = self.ocps
+        self.configs = (yaml.safe_dump(self.env_configs),)
 
 
 def run_obssys_standby_disabled() -> None:
-    script_class = ObsSysStandbyDisabled()
-    num_scripts = len(script_class.scripts)
-    print(f"\nObsSys Standby to Disabled; running {num_scripts} scripts")
-    asyncio.run(script_class.run())
+    # Ensure the invocation is correct.
+    # If not, raise KeyError.
+    # If it is correct, execute the state transition.
+    args = get_test_env_arg()
+    try:
+        script_class = ObsSysStandbyDisabled(
+            test_env=args.test_env,
+        )
+    except KeyError as ke:
+        print(repr(ke))
+    else:
+        num_scripts = len(script_class.scripts)
+        print(
+            f"\nObsSys Standby to Disabled; "
+            f"running {num_scripts} scripts "
+            f"on the '{args.test_env}' environment. "
+            f"with this configuration: \n"
+            f"{script_class.configs}"
+        )
+        asyncio.run(script_class.run())

python/lsst/ts/IntegrationTests/utils.py

@@ -0,0 +1,41 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+# This file is part of ts_IntegrationTests.
+#
+# Developed for the Vera C. Rubin Observatory Telescope & Site Software system.
+# This product includes software developed by the Vera C. Rubin Observatory
+# Project (https://www.lsst.org).
+# See the COPYRIGHT file at the top-level directory of this distribution
+# for details of code ownership.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+
+import argparse
+
+
+def get_test_env_arg() -> None:
+    # Define the script arguments.
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "test_env",
+        nargs="?",
+        type=str.lower,
+        choices=["bts", "tts", "summit"],
+        help="Specify on which environment the tests are running (case insensitive).",
+    )
+    args = parser.parse_args()
+    # Print the help if the environment is not defined.
+    if not (args.test_env):
+        parser.print_help()
+        exit()
+    return args

tests/test_enabled_offline.py

@@ -50,7 +50,7 @@ async def test_enabled_offline(self) -> None:
 
         """
         # Instantiate the EnabledOffline integration tests.
-        script_class = EnabledOffline()
+        script_class = EnabledOffline(test_env="bts")
         # Get number of scripts
         num_scripts = len(script_class.scripts)
         print(f"Enabled to Offline; running {num_scripts} scripts")

tests/test_load_camera_playlist.py

@@ -22,7 +22,6 @@
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
 import subprocess
-import sys
 import unittest
 
 from lsst.ts import salobj
@@ -55,7 +54,6 @@ async def test_camera_playlist(self) -> None:
         # script expects.
         test_camera = "at"
         test_playlist = "test"
-        sys.argv[1:] = [test_camera, test_playlist]
         # Instantiate the LoadCameraPlaylist integration tests.
         script_class = LoadCameraPlaylist(
             camera=test_camera, playlist_shortname=test_playlist

tests/test_obssys_state_transitions.py

@@ -23,6 +23,7 @@
 
 import unittest
 
+import yaml
 from lsst.ts import salobj
 from lsst.ts.IntegrationTests import (
     ObsSysDisabledEnabled,
@@ -49,38 +50,52 @@ async def test_obssys_standby_disabled(self) -> None:
         which runs the ts_standardscripts/set_summary_state.py script.
         Use the configuration stored in the obssys_state_transition_configs.py
         module.
-
         """
+
         # Instantiate the ObsSysStandbyDisabled integration tests.
-        script_class = ObsSysStandbyDisabled()
-        # Get number of scripts
+        script_class = ObsSysStandbyDisabled(test_env="tts")
+        # Get number of scripts and the configuration.
         num_scripts = len(script_class.scripts)
-        print(f"ObsSys Standby to Disabled; running {num_scripts} scripts")
+        script_config = yaml.safe_load(script_class.configs[0])
+        print(
+            f"ObsSys Standby to Disabled; running {num_scripts} scripts"
+            f" on the TTS environment, with this configuration: \n"
+            f"{script_config}"
+        )
         # Execute the scripts.
         await script_class.run()
         # Assert script was added to ScriptQueue.
         self.assertEqual(len(self.controller.queue_list), num_scripts)
         # Assert scripts passed.
         self.assertEqual(script_class.script_states, [8])
+        # Assert OCPS index was set correctly.
+        self.assertEqual(script_config["data"][3][0], "OCPS:2")
 
     async def test_obssys_disabled_enabled(self) -> None:
         """Execute the ObsSysDisabledEnabled integration test script,
         which runs the ts_standardscripts/set_summary_state.py script.
         Use the configuration stored in the obssys_state_transition_configs.py
         module.
-
         """
+
         # Instantiate the ObsSysDisabledEnabled integration tests.
-        script_class = ObsSysDisabledEnabled()
-        # Get number of scripts
+        script_class = ObsSysDisabledEnabled(test_env="bts")
+        # Get number of scripts and the configuration.
         num_scripts = len(script_class.scripts)
-        print(f"ObsSys Disabled to Enabled; running {num_scripts} scripts")
+        script_config = yaml.safe_load(script_class.configs[0])
+        print(
+            f"ObsSys Disabled to Enabled; running {num_scripts} scripts"
+            f" on the BTS environment, with this configuration: \n"
+            f"{script_config}"
+        )
         # Execute the scripts.
         await script_class.run()
         # Assert script was added to ScriptQueue.
         self.assertEqual(len(self.controller.queue_list), num_scripts)
         # Assert scripts passed.
         self.assertEqual(script_class.script_states, [8])
+        # Assert OCPS index was set correctly.
+        self.assertEqual(script_config["data"][3][0], "OCPS:3")
 
     async def asyncTearDown(self) -> None:
         await self.controller.close()