Skip to content

Add system tests for default handling + ruff + pyright #135

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
LowriJenkins wants to merge 3 commits into
base: master
Choose a base branch
from
Open

Add system tests for default handling + ruff + pyright #135

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
c9a046f
Add system tests for default handling + ruff + pyright
LowriJenkins Oct 1, 2025
bf1310f
Merge branch 'master' into Ticket8445_Add_system_tests_for_new_defaul…
LowriJenkins Jan 7, 2026
eb9a361
Fix failing tests
LowriJenkins Jan 21, 2026
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
186 changes: 165 additions & 21 deletions test_blockserver.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -27,6 +28,34 @@ 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
Expand All @@ -48,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)
Expand All @@ -62,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)
Expand Down Expand Up @@ -114,13 +147,17 @@ 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(
[
("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",
Expand Down Expand Up @@ -157,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
]
Expand Down Expand Up @@ -303,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(
Expand All @@ -321,24 +366,31 @@ 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)

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):
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:
Expand All @@ -352,16 +404,18 @@ 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):
CONFIGURATION_NAME = "test_blockserver_save_active_config"
BLOCK_NAME = "TEST_BLOCK"
utilities.load_config_if_not_already_loaded(CONFIGURATION_NAME)
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)

data = {}
data["name"] = CONFIGURATION_NAME
data["name"] = configuration_name
data["blocks"] = [
{
"name": BLOCK_NAME,
"name": block_name,
"pv": "TEST_PV",
}
]
Expand All @@ -373,4 +427,94 @@ def test_WHEN_block_is_added_to_active_config_via_save_new_config_pv_THEN_block_
is_local=True,
)
wait_for_server()
self.assertTrue(utilities.check_block_exists(BLOCK_NAME))
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",
compress_and_hex(json.dumps(data)),
wait=True,
is_local=True,
)
wait_for_server()
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)
Loading