From c9a046f8a82801816837cb4f6e5a68161e2355b0 Mon Sep 17 00:00:00 2001 From: Lowri Jenkins Date: Wed, 1 Oct 2025 11:45:26 +0100 Subject: [PATCH 1/2] Add system tests for default handling + ruff + pyright --- test_blockserver.py | 149 +++++++++++++++++++++++++++++++++++++++----- 1 file changed, 135 insertions(+), 14 deletions(-) diff --git a/test_blockserver.py b/test_blockserver.py index c2effdb..a710b54 100644 --- a/test_blockserver.py +++ b/test_blockserver.py @@ -6,8 +6,9 @@ import requests from genie_python import genie as g +from genie_python.channel_access_exceptions import UnableToConnectToPVException from genie_python.utilities import compress_and_hex -from hamcrest import * +from hamcrest import assert_that, contains_string, ends_with, is_ from parameterized import parameterized from utilities import utilities @@ -27,13 +28,41 @@ def wait_for_server(): time.sleep(1) +def test_pv_with_macro_value(macro_value, pv, test_case): + if macro_value == "": + test_case.assertRaises(UnableToConnectToPVException, g.get_pv, pv, is_local=True) + else: + test_case.assertEqual( + g.get_pv( + pv, + is_local=True, + ), + macro_value, + ) + + +def test_config_macros( + use_default, config_macro, default_value, macro_value, macro_name, test_case +): + if use_default: + test_case.assertNotIn(macro_name, config_macro) + return default_value + else: + test_case.assertIn(macro_name, config_macro) + test_case.assertEqual(macro_value, config_macro[macro_name]["value"]) + test_case.assertNotIn("useDefault", config_macro[macro_name]) + if macro_value == "": + return default_value + return macro_value + + class TestBlockserver(unittest.TestCase): """ Tests for top-level functionality of block server """ def setUp(self) -> None: - g.set_instrument(None, import_instrument_init=False) + g.set_instrument("", import_instrument_init=False) self.pvlist_file = os.path.join(r"C:\Instrument", "Settings", "gwblock.pvlist") self.rc_settings_file = os.path.join( r"C:\Instrument", @@ -84,7 +113,7 @@ def test_GIVEN_config_changes_by_block_THEN_iocs_do_not_restart_except_for_caenv details["desc"] = "some_edited_description" g.set_pv( "CS:BLOCKSERVER:SET_CURR_CONFIG_DETAILS", - compress_and_hex(json.dumps(details)), + str(compress_and_hex(json.dumps(details))), is_local=True, ) @@ -114,7 +143,8 @@ def test_GIVEN_config_changes_by_block_THEN_iocs_do_not_restart_except_for_caenv err = e time.sleep(1) else: - raise err + if err is not None: + raise err @parameterized.expand( parameterized_list( @@ -333,9 +363,10 @@ def test_GIVEN_config_contains_rc_settings_THEN_configuration_rc_settings_used(s utilities.load_config_if_not_already_loaded(config) config_rc_settings_file = os.path.join(self.config_dir, config, "rc_settings.cmd") - with open(self.rc_settings_file, "r") as rc_settings, open( - config_rc_settings_file, "r" - ) as config_rc_settings: + with ( + open(self.rc_settings_file, "r") as rc_settings, + open(config_rc_settings_file, "r") as config_rc_settings, + ): assert_that(rc_settings.read(), is_(config_rc_settings.read())) def test_GIVEN_config_claims_but_does_not_contain_rc_settings_THEN_rc_settings_generated(self): @@ -353,24 +384,114 @@ def test_GIVEN_config_does_not_contain_rc_settings_THEN_rc_settings_generated(se assert_that(file_content, contains_string("$(MYPVPREFIX)CS:SB:A")) def test_WHEN_block_is_added_to_active_config_via_save_new_config_pv_THEN_block_added(self): - CONFIGURATION_NAME = "test_blockserver_save_active_config" - BLOCK_NAME = "TEST_BLOCK" - utilities.load_config_if_not_already_loaded(CONFIGURATION_NAME) + configuration_name = "test_blockserver_save_active_config" + block_name = "TEST_BLOCK" + utilities.load_config_if_not_already_loaded(configuration_name) data = {} - data["name"] = CONFIGURATION_NAME + data["name"] = configuration_name data["blocks"] = [ { - "name": BLOCK_NAME, + "name": block_name, "pv": "TEST_PV", } ] g.set_pv( "CS:BLOCKSERVER:SAVE_NEW_CONFIG", - compress_and_hex(json.dumps(data)), + str(compress_and_hex(json.dumps(data))), + wait=True, + is_local=True, + ) + wait_for_server() + self.assertTrue(utilities.check_block_exists(block_name)) + + @parameterized.expand( + parameterized_list( + [ + ("_No_Vals_All_Default", "", True, "", True, "", True), + ("_No_Vals_No_Default", "", False, "", False, "", False), + ("_Required_Vals_Else_Default", "1", False, "", True, "3", False), + ("_set_Vals_All_Default", "1", True, "2", True, "3", True), + ("_set_Vals_No_Default", "1", False, "2", False, "3", False), + ] + ) + ) + def test_WHEN_ioc_has_macros_THEN_defaults_handled_correctly( + self, + _, + name, + macro_1_val, + macro_1_use_default, + macro_2_val, + macro_2_use_default, + macro_3_val, + macro_3_use_default, + ): + configuration_name = "test_blockserver_save_active_config" + utilities.load_config_if_not_already_loaded(configuration_name) + data = {} + data["name"] = configuration_name + data["iocs"] = [ + { + "name": "SIMPLE", + "autostart": "true", + "restart": "true", + "macros": [ + { + "name": "MACRO1", + "value": f"{macro_1_val}", + "description": "A macro without a default", + "pattern": "", + "defaultValue": "", + "useDefault": f"{macro_1_use_default}", + "hasDefault": "NO", + }, + { + "name": "MACRO2", + "value": f"{macro_2_val}", + "description": "A macro with a default of 5", + "pattern": "", + "defaultValue": "", + "useDefault": f"{macro_2_use_default}", + "hasDefault": "YES", + }, + { + "name": "MACRO3", + "value": f"{macro_3_val}", + "description": "A macro with a default of ''", + "pattern": "", + "defaultValue": "", + "useDefault": f"{macro_3_use_default}", + "hasDefault": "YES", + }, + ], + } + ] + + g.set_pv( + "CS:BLOCKSERVER:SAVE_NEW_CONFIG", + str(compress_and_hex(json.dumps(data))), wait=True, is_local=True, ) wait_for_server() - self.assertTrue(utilities.check_block_exists(BLOCK_NAME)) + macros_in_config = utilities.get_config_details()["iocs"][0]["macros"] + macros_in_config = {macro["name"]: macro for macro in macros_in_config} + expected_1 = test_config_macros( + macro_1_use_default, macros_in_config, "", macro_1_val, "MACRO1", self + ) + expected_2 = test_config_macros( + macro_2_use_default, macros_in_config, "5", macro_2_val, "MACRO2", self + ) + expected_3 = test_config_macros( + macro_3_use_default, macros_in_config, "", macro_3_val, "MACRO3", self + ) + + assert_with_timeout( + lambda: self.assertTrue(utilities.is_ioc_up("SIMPLE")), + timeout=SECONDS_TO_WAIT_FOR_IOC_STARTS, + ) + test_pv_with_macro_value(expected_1, "SIMPLE:MACROTEST1", self) + test_pv_with_macro_value(expected_2, "SIMPLE:MACROTEST2", self) + test_pv_with_macro_value(expected_3, "SIMPLE:MACROTEST3", self) From eb9a361833221c22fd8fc4b9a2ef61812251e24a Mon Sep 17 00:00:00 2001 From: Lowri Jenkins Date: Wed, 21 Jan 2026 10:47:34 +0000 Subject: [PATCH 2/2] Fix failing tests --- test_blockserver.py | 51 ++++++++++++++++++++++++++++++++------------- 1 file changed, 37 insertions(+), 14 deletions(-) diff --git a/test_blockserver.py b/test_blockserver.py index a710b54..04a5b17 100644 --- a/test_blockserver.py +++ b/test_blockserver.py @@ -62,7 +62,7 @@ class TestBlockserver(unittest.TestCase): """ def setUp(self) -> None: - g.set_instrument("", import_instrument_init=False) + g.set_instrument(None, import_instrument_init=False) self.pvlist_file = os.path.join(r"C:\Instrument", "Settings", "gwblock.pvlist") self.rc_settings_file = os.path.join( r"C:\Instrument", @@ -77,7 +77,9 @@ def setUp(self) -> None: ) self.block_archive_blocks_url = "http://localhost:4813/group?name=BLOCKS&format=json" - def test_GIVEN_config_changes_THEN_dae_and_instetc_come_back_with_autorestart_reapplied(self): + def test_GIVEN_config_changes_THEN_dae_and_instetc_come_back_with_autorestart_reapplied( + self, + ): g.reload_current_config() iocs_to_check = ["ISISDAE_01", "INSTETC_01"] utilities.wait_for_iocs_to_be_up(iocs_to_check, SECONDS_TO_WAIT_FOR_IOC_STARTS) @@ -91,7 +93,9 @@ def test_GIVEN_config_changes_THEN_dae_and_instetc_come_back_with_autorestart_re self.assertEqual(status.lower(), "running") self.assertEqual(autorestart.lower(), "on") - def test_GIVEN_config_changes_by_block_THEN_iocs_do_not_restart_except_for_caenv895(self): + def test_GIVEN_config_changes_by_block_THEN_iocs_do_not_restart_except_for_caenv895( + self, + ): utilities.load_config_if_not_already_loaded("test_blockserver") utilities.wait_for_iocs_to_be_up(["SIMPLE", "CAENV895_01"], SECONDS_TO_WAIT_FOR_IOC_STARTS) @@ -113,7 +117,7 @@ def test_GIVEN_config_changes_by_block_THEN_iocs_do_not_restart_except_for_caenv details["desc"] = "some_edited_description" g.set_pv( "CS:BLOCKSERVER:SET_CURR_CONFIG_DETAILS", - str(compress_and_hex(json.dumps(details))), + compress_and_hex(json.dumps(details)), is_local=True, ) @@ -150,7 +154,10 @@ def test_GIVEN_config_changes_by_block_THEN_iocs_do_not_restart_except_for_caenv parameterized_list( [ ("simple1", "simple2"), # Move IOC between 2 configs - ("simple_comp_macros", "simple_comp_macros_2"), # Move IOC between two components + ( + "simple_comp_macros", + "simple_comp_macros_2", + ), # Move IOC between two components ( "simple_with_macros", "simple_comp_macros", @@ -187,7 +194,10 @@ def test_GIVEN_config_changes_by_ioc_and_ioc_has_same_settings_in_old_and_new_WH parameterized_list( [ ("simple1", "simple_with_macros"), # Config -> config - ("simple_comp_no_macros", "simple_comp_macros"), # Component -> component + ( + "simple_comp_no_macros", + "simple_comp_macros", + ), # Component -> component ("simple1", "simple_comp_macros"), # Config -> Component ("simple_comp_macros", "simple1"), # Component -> Config ] @@ -333,12 +343,17 @@ def test_GIVEN_config_does_not_contain_gw_and_archiver_files_THEN_archiver_confi assert_that(len(response["Channels"]), is_(1)) assert_that(response["Channels"][0]["Channel"], ends_with("a")) - def test_GIVEN_config_contains_gw_and_archiver_files_THEN_configuration_pvlist_used(self): + def test_GIVEN_config_contains_gw_and_archiver_files_THEN_configuration_pvlist_used( + self, + ): config = "test_blockserver_with_gw_archiver" utilities.load_config_if_not_already_loaded(config) config_pvlist_file = os.path.join(self.config_dir, config, "gwblock.pvlist") - with open(self.pvlist_file, "r") as pvlist, open(config_pvlist_file, "r") as config_pvlist: + with ( + open(self.pvlist_file, "r") as pvlist, + open(config_pvlist_file, "r") as config_pvlist, + ): assert_that(pvlist.read(), is_(config_pvlist.read())) def test_GIVEN_config_claims_but_does_not_contain_gw_and_archiver_files_THEN_pvlist_generated( @@ -351,14 +366,18 @@ def test_GIVEN_config_claims_but_does_not_contain_gw_and_archiver_files_THEN_pvl assert_that(file_content, contains_string("TIZRWARNING")) assert_that(file_content, contains_string("TIZROUTOFRANGE")) - def test_GIVEN_config_does_not_contain_gw_and_archiver_files_THEN_pvlist_generated(self): + def test_GIVEN_config_does_not_contain_gw_and_archiver_files_THEN_pvlist_generated( + self, + ): utilities.load_config_if_not_already_loaded("test_blockserver") with open(self.pvlist_file, "r") as pvlist: file_content = pvlist.read() assert_that(file_content, contains_string(g.prefix_pv_name("CS:SB:a"))) - def test_GIVEN_config_contains_rc_settings_THEN_configuration_rc_settings_used(self): + def test_GIVEN_config_contains_rc_settings_THEN_configuration_rc_settings_used( + self, + ): config = "test_blockserver_with_gw_archiver" utilities.load_config_if_not_already_loaded(config) @@ -369,7 +388,9 @@ def test_GIVEN_config_contains_rc_settings_THEN_configuration_rc_settings_used(s ): assert_that(rc_settings.read(), is_(config_rc_settings.read())) - def test_GIVEN_config_claims_but_does_not_contain_rc_settings_THEN_rc_settings_generated(self): + def test_GIVEN_config_claims_but_does_not_contain_rc_settings_THEN_rc_settings_generated( + self, + ): utilities.load_config_if_not_already_loaded("test_blockserver_without_gw_archiver") with open(self.rc_settings_file, "r") as rc_settings: @@ -383,7 +404,9 @@ def test_GIVEN_config_does_not_contain_rc_settings_THEN_rc_settings_generated(se file_content = rc_settings.read() assert_that(file_content, contains_string("$(MYPVPREFIX)CS:SB:A")) - def test_WHEN_block_is_added_to_active_config_via_save_new_config_pv_THEN_block_added(self): + def test_WHEN_block_is_added_to_active_config_via_save_new_config_pv_THEN_block_added( + self, + ): configuration_name = "test_blockserver_save_active_config" block_name = "TEST_BLOCK" utilities.load_config_if_not_already_loaded(configuration_name) @@ -399,7 +422,7 @@ def test_WHEN_block_is_added_to_active_config_via_save_new_config_pv_THEN_block_ g.set_pv( "CS:BLOCKSERVER:SAVE_NEW_CONFIG", - str(compress_and_hex(json.dumps(data))), + compress_and_hex(json.dumps(data)), wait=True, is_local=True, ) @@ -471,7 +494,7 @@ def test_WHEN_ioc_has_macros_THEN_defaults_handled_correctly( g.set_pv( "CS:BLOCKSERVER:SAVE_NEW_CONFIG", - str(compress_and_hex(json.dumps(data))), + compress_and_hex(json.dumps(data)), wait=True, is_local=True, )