diff --git a/archivist/compliance_policies.py b/archivist/compliance_policies.py index ed61da25..0683580b 100644 --- a/archivist/compliance_policies.py +++ b/archivist/compliance_policies.py @@ -167,7 +167,7 @@ def count(self, *, props: Optional[dict[str, Any]] = None) -> int: Counts number of compliance policies that match criteria. Args: - props (dict): e.g. {"display_name": "foo" } + props (dict): e.g. {"compliance_type": "COMPLIANCE_RICHNESS" } Returns: integer count of compliance policies. diff --git a/archivist/compliance_policy_requests.py b/archivist/compliance_policy_requests.py index 8306b744..3216e114 100644 --- a/archivist/compliance_policy_requests.py +++ b/archivist/compliance_policy_requests.py @@ -50,6 +50,7 @@ class CompliancePolicyCurrentOutstanding(CompliancePolicyBase): """ event_display_type: str + closing_event_display_type: str compliance_type: str = CompliancePolicyType.COMPLIANCE_CURRENT_OUTSTANDING.name diff --git a/docs/compliance_policies_since.rst b/docs/compliance_policies_since.rst new file mode 100644 index 00000000..4623d7ed --- /dev/null +++ b/docs/compliance_policies_since.rst @@ -0,0 +1,10 @@ +.. _compliance_policies_sinceref: + +Since Compliance Policy +........................ + +.. literalinclude:: ../examples/compliance_policies_since.py + :language: python + + + diff --git a/docs/getting_started.rst b/docs/getting_started.rst index 6fff2037..93951556 100644 --- a/docs/getting_started.rst +++ b/docs/getting_started.rst @@ -25,6 +25,8 @@ See the examples and functests directories. access_policy_create access_policies_filter + compliance_policies_since + runner/index scan_test diff --git a/examples/compliance_policies_since.py b/examples/compliance_policies_since.py new file mode 100644 index 00000000..0e53fd4f --- /dev/null +++ b/examples/compliance_policies_since.py @@ -0,0 +1,157 @@ +"""Define a compliance policy that alerts when an asset has expired. + +Main function parses in a url to the Archivist and client credentials , which is +a user authorization. The main function would initialize an archivist connection +using the url and the credentials, called "arch", then call arch.access_policies.list() +with suitable properties and attributes. + +""" +from json import dumps as json_dumps +from os import getenv +from time import sleep +from uuid import uuid4 +from warnings import filterwarnings + +from archivist.archivist import Archivist +from archivist.compliance_policy_requests import ( + CompliancePolicySince, +) +from archivist.utils import get_auth + +filterwarnings("ignore", message="Unverified HTTPS request") + + +def get_archivist(): + """Create Archivist endpoint.""" + + # client id and client secret is obtained from the appidp endpoint - see the + # application registrations example code in examples/applications_registration.py + # + # client id is an environment variable. client_secret is stored in a file in a + # directory that has 0700 permissions. The location of this file is set in + # the client_secret_file environment variable. + # + auth = get_auth( + auth_token=getenv("ARCHIVIST_AUTHTOKEN"), + auth_token_filename=getenv("ARCHIVIST_AUTHTOKEN_FILENAME"), + client_id=getenv("ARCHIVIST_CLIENT_ID"), + client_secret=getenv("ARCHIVIST_CLIENT_SECRET"), + client_secret_filename=getenv("ARCHIVIST_CLIENT_SECRET_FILENAME"), + ) + + # Initialize connection to Archivist + arch = Archivist( + "https://app.rkvst.io", + auth, + ) + return arch + + +def create_compliance_policy(arch, tag): + """Compliance policy which expires 10 seconds after a + Maintenance Performed event on a 'Traffic Light' has occurred. + + Usually the expiry time is on the order of days or weeks.. + + Additionally the use of tag is simply to make this example + repeatable. + """ + compliance_policy = arch.compliance_policies.create( + CompliancePolicySince( + description="Maintenance should be performed every 10 seconds", + display_name="Regular Maintenance of Traffic light", + asset_filter=[ + ["attributes.arc_display_type=Traffic Light"], + ], + event_display_type=f"Maintenance Performed {tag}", + time_period_seconds=10, # very short so we can test + ) + ) + print("SINCE_POLICY:", json_dumps(compliance_policy, indent=4)) + return compliance_policy + + +def create_traffic_light(arch): + """ + Creates a traffic light. + + Note that arc_display_type siginfies a Traffic Light + """ + + traffic_light = arch.assets.create( + attrs={ + "arc_display_name": "Traffic light model 54", + "arc_description": "Traffic flow control light at A603 North East", + "arc_display_type": "Traffic Light", + }, + confirm=True, + ) + print("TRAFFIC_LIGHT:", json_dumps(traffic_light, indent=4)) + return traffic_light + + +def perform_maintenance(arch, traffic_light, tag): + """ + Perform maintenance on traffic light + """ + maintenance_performed = arch.events.create( + traffic_light["identity"], + { + "operation": "Record", + "behaviour": "RecordEvidence", + }, + { + "arc_description": "Maintenance performed on traffic light", + "arc_display_type": f"Maintenance Performed {tag}", + }, + confirm=True, + ) + print("MAINTENANCE_PERFORMED:", json_dumps(maintenance_performed, indent=4)) + + +def main(): + """ + Connect to archivist, create an asset, create a compliance policy + execute an event on the asset and check if the asset has expired + """ + # first get Archivist connection. + arch = get_archivist() + + tag = uuid4() # make this example repeatable + + # make a SINCE compliance policy that alerts when the + # maintenance performed event has expired. + compliance_policy = create_compliance_policy(arch, tag) + + # create an asset that matches the assets_filter field in the + # compliance policy. + traffic_light = create_traffic_light(arch) + + # perform maintenance on the asset which is valid for 10 seconds. + perform_maintenance(arch, traffic_light, tag) + + # and check compliance - should be OK. + print("Sleep 1 second...") + sleep(1) + compliance = arch.compliance.compliant_at( + traffic_light["identity"], + ) + print("COMPLIANCE (true):", json_dumps(compliance, indent=4)) + + # however waiting long enough (> 10s) will cause the asset to + # become non-compliant... + print("Sleep 15 seconds...") + sleep(15) + compliance = arch.compliance.compliant_at( + traffic_light["identity"], + ) + print("COMPLIANCE (false):", json_dumps(compliance, indent=4)) + + # finally delete the compliance_policy + arch.compliance_policies.delete( + compliance_policy["identity"], + ) + + +if __name__ == "__main__": + main() diff --git a/functests/execcompliance_policies.py b/functests/execcompliance_policies.py new file mode 100644 index 00000000..c0531ae9 --- /dev/null +++ b/functests/execcompliance_policies.py @@ -0,0 +1,388 @@ +""" +Test compliance policies +""" +from json import dumps as json_dumps +from os import getenv +from time import sleep +from unittest import TestCase +from uuid import uuid4 + +from archivist.archivist import Archivist +from archivist.compliance_policy_requests import ( + CompliancePolicyCurrentOutstanding, + CompliancePolicyDynamicTolerance, + CompliancePolicyPeriodOutstanding, + CompliancePolicyRichness, + CompliancePolicySince, +) +from archivist.compliance_policy_type import CompliancePolicyType +from archivist.utils import get_auth + +# pylint: disable=fixme +# pylint: disable=missing-docstring +# pylint: disable=unused-variable + +from archivist import logger + +if getenv("TEST_DEBUG") is not None: + logger.set_logger(getenv("TEST_DEBUG")) +else: + logger.set_logger("INFO") + +LOGGER = logger.LOGGER + +# Ridiculaously short maintenance period for test purposes +SINCE_POLICY = CompliancePolicySince( + description="Maintenance should be performed every 10 seconds", + display_name="Regular Maintenance of Traffic light", + asset_filter=[ + ["attributes.arc_display_type=Traffic Light"], + ], + event_display_type="Maintenance Performed", + time_period_seconds=10, # very short so we can test +) + +CURRENT_OUTSTANDING_POLICY = CompliancePolicyCurrentOutstanding( + description="Maintenance should be performed every 10 seconds", + display_name="Regular Maintenance of Traffic light", + asset_filter=[ + ["attributes.arc_display_type=Traffic Light"], + ], + event_display_type="Maintenance Request", + closing_event_display_type="Maintenance Performed", +) + +PERIOD_OUTSTANDING_POLICY = CompliancePolicyPeriodOutstanding( + description="period_outstanding description", + display_name="period_outstanding display_name", + asset_filter=[ + ["attributes.radioactive=true"], + ], + event_display_type="period_outstanding event_display_type", + closing_event_display_type="period_outstanding closing_event_display_type", + time_period_seconds=10, +) +DYNAMIC_TOLERANCE_POLICY = CompliancePolicyDynamicTolerance( + description="dynamic_tolerance description", + display_name="dynamic_tolerance display_name", + asset_filter=[ + ["attributes.radioactive=true"], + ], + event_display_type="dynamic_tolerance event_display_type", + closing_event_display_type="dynamic_tolerance closing_event_display_type", + dynamic_window=86400, + dynamic_variability=0.5, +) +RICHNESS_POLICY = CompliancePolicyRichness( + description="richness description", + display_name="richness display_name", + asset_filter=[ + ["attributes.radioactive=true"], + ], + richness_assertions=[ + ["rad<7"], + ], +) + + +class TestCompliancePoliciesBase(TestCase): + """ + Test Archivist CompliancePolicies Create method + """ + + maxDiff = None + + def setUp(self): + auth = get_auth( + auth_token_filename=getenv("TEST_AUTHTOKEN_FILENAME"), + client_id=getenv("TEST_CLIENT_ID"), + client_secret=getenv("TEST_CLIENT_SECRET"), + client_secret_filename=getenv("TEST_CLIENT_SECRET_FILENAME"), + ) + self.arch = Archivist(getenv("TEST_ARCHIVIST"), auth, verify=False) + print() + + +class TestCompliancePolicies(TestCompliancePoliciesBase): + def test_compliancepolicies_create_since(self): + """ + Test compliance_policies creation + """ + compliance_policy = self.arch.compliance_policies.create( + SINCE_POLICY, + ) + self.assertEqual( + compliance_policy["display_name"], + SINCE_POLICY.display_name, + msg="Incorrect display name", + ) + print("SINCE_POLICY:", json_dumps(compliance_policy, indent=4)) + self.arch.compliance_policies.delete( + compliance_policy["identity"], + ) + + def test_compliancepolicies_create_richness(self): + """ + Test compliance_policies creation + """ + compliance_policy = self.arch.compliance_policies.create( + RICHNESS_POLICY, + ) + self.assertEqual( + compliance_policy["display_name"], + RICHNESS_POLICY.display_name, + msg="Incorrect display name", + ) + print("RICHNESS_POLICY:", json_dumps(compliance_policy, indent=4)) + self.arch.compliance_policies.delete( + compliance_policy["identity"], + ) + + def test_compliancepolicies_create_dynamic_tolerance(self): + """ + Test compliance_policies creation + """ + compliance_policy = self.arch.compliance_policies.create( + DYNAMIC_TOLERANCE_POLICY, + ) + self.assertEqual( + compliance_policy["display_name"], + DYNAMIC_TOLERANCE_POLICY.display_name, + msg="Incorrect display name", + ) + print("DYNAMIC_TOLERANCE_POLICY:", json_dumps(compliance_policy, indent=4)) + self.arch.compliance_policies.delete( + compliance_policy["identity"], + ) + + def test_compliancepolicies_create_current_outstanding(self): + """ + Test compliance_policies creation + """ + compliance_policy = self.arch.compliance_policies.create( + CURRENT_OUTSTANDING_POLICY, + ) + self.assertEqual( + compliance_policy["display_name"], + CURRENT_OUTSTANDING_POLICY.display_name, + msg="Incorrect display name", + ) + print("CURRENT_OUTSTANDING_POLICY:", json_dumps(compliance_policy, indent=4)) + self.arch.compliance_policies.delete( + compliance_policy["identity"], + ) + + def test_compliancepolicies_create_period_understanding(self): + """ + Test compliance_policies creation + """ + compliance_policy = self.arch.compliance_policies.create( + PERIOD_OUTSTANDING_POLICY, + ) + self.assertEqual( + compliance_policy["display_name"], + PERIOD_OUTSTANDING_POLICY.display_name, + msg="Incorrect display name", + ) + print("PERIOD_OUTSTANDING_POLICY:", json_dumps(compliance_policy, indent=4)) + self.arch.compliance_policies.delete( + compliance_policy["identity"], + ) + + def test_compliance_policies_list(self): + """ + Test compliance_policy list + """ + compliance_policies = list(self.arch.compliance_policies.list()) + for i, compliance_policy in enumerate(compliance_policies): + print(i, ":", json_dumps(compliance_policy, indent=4)) + self.assertGreater( + len(compliance_policy["display_name"]), + 0, + msg="Incorrect display name", + ) + self.arch.compliance_policies.delete( + compliance_policy["identity"], + ) + + def test_compliance_policies_count(self): + """ + Test compliance_policy count + """ + count = self.arch.compliance_policies.count( + props={"compliance_type": CompliancePolicyType.COMPLIANCE_SINCE.name} + ) + print("No. of 'SINCE' compliance policies:", count) + count = self.arch.compliance_policies.count( + props={"compliance_type": CompliancePolicyType.COMPLIANCE_RICHNESS.name} + ) + print("No. of 'RICHNESS' compliance policies:", count) + count = self.arch.compliance_policies.count( + props={ + "compliance_type": CompliancePolicyType.COMPLIANCE_DYNAMIC_TOLERANCE.name + } + ) + print("No. of 'DYNAMIC_TOLERANCE' compliance policies:", count) + count = self.arch.compliance_policies.count( + props={ + "compliance_type": CompliancePolicyType.COMPLIANCE_CURRENT_OUTSTANDING.name + } + ) + print("No. of 'CURRENT_OUTSTANDING' compliance policies:", count) + count = self.arch.compliance_policies.count( + props={ + "compliance_type": CompliancePolicyType.COMPLIANCE_PERIOD_OUTSTANDING.name + } + ) + print("No. of 'PERIOD_OUTSTANDING' compliance policies:", count) + + +TRAFFIC_LIGHT = { + "arc_display_name": "Traffic light model 54", + "arc_description": "Traffic flow control light at A603 North East", + "arc_display_type": "Traffic Light", +} + +PROPS = { + "operation": "Record", + "behaviour": "RecordEvidence", +} + + +class TestCompliancePoliciesCompliantAt(TestCompliancePoliciesBase): + def test_compliancepolicies_since(self): + """ + Test compliance_policies creation + """ + tag = uuid4() + compliance_policy = self.arch.compliance_policies.create( + CompliancePolicySince( + description="Maintenance should be performed every 10 seconds", + display_name="Regular Maintenance of Traffic light", + asset_filter=[ + ["attributes.arc_display_type=Traffic Light"], + ], + event_display_type=f"Maintenance Performed {tag}", + time_period_seconds=10, # very short so we can test + ) + ) + print("SINCE_POLICY:", json_dumps(compliance_policy, indent=4)) + + traffic_light = self.arch.assets.create( + attrs=TRAFFIC_LIGHT, + confirm=True, + ) + print("TRAFFIC_LIGHT:", json_dumps(traffic_light, indent=4)) + + maintenance_performed = self.arch.events.create( + traffic_light["identity"], + PROPS, + { + "arc_description": "Maintenance performed on traffic light", + "arc_display_type": f"Maintenance Performed {tag}", + }, + confirm=True, + ) + print("MAINTENANCE_PERFORMED:", json_dumps(maintenance_performed, indent=4)) + + print("Sleep 1 second...") + sleep(1) + compliance = self.arch.compliance.compliant_at( + traffic_light["identity"], + ) + print("COMPLIANCE (true):", json_dumps(compliance, indent=4)) + self.assertTrue( + compliance["compliant"], + msg="Assets should be compliant", + ) + + print("Sleep 15 seconds...") + sleep(15) + compliance = self.arch.compliance.compliant_at( + traffic_light["identity"], + ) + print("COMPLIANCE (false):", json_dumps(compliance, indent=4)) + self.assertFalse( + compliance["compliant"], + msg="Assets should not be compliant", + ) + + compliance_policy = self.arch.compliance_policies.delete( + compliance_policy["identity"], + ) + + def test_compliancepolicies_current_outstanding(self): + """ + Test compliance_policies creation + """ + tag = uuid4() + compliance_policy = self.arch.compliance_policies.create( + CompliancePolicyCurrentOutstanding( + description="Maintenance should be completed", + display_name="Regular Maintenance of Traffic light", + asset_filter=[ + ["attributes.arc_display_type=Traffic Light"], + ], + event_display_type=f"Maintenance Request {tag}", + closing_event_display_type=f"Maintenance Performed {tag}", + ), + ) + print("CURRENT_OUTSTANDING_POLICY:", json_dumps(compliance_policy, indent=4)) + + traffic_light = self.arch.assets.create( + attrs=TRAFFIC_LIGHT, + confirm=True, + ) + print("TRAFFIC_LIGHT:", json_dumps(traffic_light, indent=4)) + + maintenance_request = self.arch.events.create( + traffic_light["identity"], + PROPS, + { + "arc_description": "Maintenance request on traffic light", + "arc_display_type": f"Maintenance Request {tag}", + "arc_correlation_value": str(tag), + }, + confirm=True, + ) + print("MAINTENANCE_REQUIRED:", json_dumps(maintenance_request, indent=4)) + + print("Sleep 1 second...") + sleep(1) + + compliance = self.arch.compliance.compliant_at( + traffic_light["identity"], + ) + print("COMPLIANCE (false):", json_dumps(compliance, indent=4)) + self.assertFalse( + compliance["compliant"], + msg="Assets should not be compliant", + ) + + maintenance_performed = self.arch.events.create( + traffic_light["identity"], + PROPS, + { + "arc_description": "Maintenance performed on traffic light", + "arc_display_type": f"Maintenance Performed {tag}", + "arc_correlation_value": str(tag), + }, + confirm=True, + ) + print("MAINTENANCE_PERFORMED:", json_dumps(maintenance_performed, indent=4)) + + print("Sleep 1 second...") + sleep(1) + + compliance = self.arch.compliance.compliant_at( + traffic_light["identity"], + ) + print("COMPLIANCE (true):", json_dumps(compliance, indent=4)) + self.assertTrue( + compliance["compliant"], + msg="Assets should be compliant", + ) + + compliance_policy = self.arch.compliance_policies.delete( + compliance_policy["identity"], + ) diff --git a/notebooks/Check Asset Compliance using SINCE Policy.ipynb b/notebooks/Check Asset Compliance using SINCE Policy.ipynb new file mode 100644 index 00000000..38b0eb22 --- /dev/null +++ b/notebooks/Check Asset Compliance using SINCE Policy.ipynb @@ -0,0 +1,268 @@ +{ + "cells": [ + { + "cell_type": "code", + "execution_count": null, + "id": "c85b9e96", + "metadata": {}, + "outputs": [], + "source": [ + "# Define a compliance policy that alerts when an asset has expired.\n", + "\n", + "# Main function parses in a url to the Archivist and client credentials , which is\n", + "# a user authorization. The main function would initialize an archivist connection\n", + "# using the url and the credentials, called \"arch\", then call arch.access_policies.list()\n", + "# with suitable properties and attributes." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "001b25e3", + "metadata": {}, + "outputs": [], + "source": [ + "from json import dumps as json_dumps\n", + "from os import getenv\n", + "from time import sleep\n", + "from uuid import uuid4\n", + "from warnings import filterwarnings\n", + "\n", + "from archivist.archivist import Archivist\n", + "from archivist.compliance_policy_requests import (\n", + " CompliancePolicySince,\n", + ")\n", + "from archivist.logger import set_logger\n", + "\n", + "filterwarnings(\"ignore\", message=\"Unverified HTTPS request\")" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "a877ffed", + "metadata": {}, + "outputs": [], + "source": [ + "URL=\"https://app.rkvst.io\"\n", + "URL=\"https://app.dev-paul-0.wild.jitsuin.io\"" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "acdf240c", + "metadata": {}, + "outputs": [], + "source": [ + "def create_compliance_policy(arch, tag):\n", + " \"\"\"Compliance policy which expires 10 seconds after a\n", + " Maintenance Performed event on a 'Traffic Light' has occurred.\n", + "\n", + " Usually the expiry time is on the order of days or weeks..\n", + "\n", + " Additionally the use of tag is simply to make this example\n", + " repeatable.\n", + " \"\"\"\n", + " compliance_policy = arch.compliance_policies.create(\n", + " CompliancePolicySince(\n", + " description=\"Maintenance should be performed every 10 seconds\",\n", + " display_name=\"Regular Maintenance of Traffic light\",\n", + " asset_filter=[\n", + " [\"attributes.arc_display_type=Traffic Light\"],\n", + " ],\n", + " event_display_type=f\"Maintenance Performed {tag}\",\n", + " time_period_seconds=10, # very short so we can test\n", + " )\n", + " )\n", + " print(\"SINCE_POLICY:\", json_dumps(compliance_policy, indent=4))\n", + " return compliance_policy" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "8889c68d", + "metadata": {}, + "outputs": [], + "source": [ + "def create_traffic_light(arch):\n", + " \"\"\"\n", + " Creates a traffic light.\n", + "\n", + " Note that arc_display_type siginfies a Traffic Light\n", + " \"\"\"\n", + "\n", + " traffic_light = arch.assets.create(\n", + " attrs={\n", + " \"arc_display_name\": \"Traffic light model 54\",\n", + " \"arc_description\": \"Traffic flow control light at A603 North East\",\n", + " \"arc_display_type\": \"Traffic Light\",\n", + " },\n", + " confirm=True,\n", + " )\n", + " print(\"TRAFFIC_LIGHT:\", json_dumps(traffic_light, indent=4))\n", + " return traffic_light" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "ba24d143", + "metadata": {}, + "outputs": [], + "source": [ + "def perform_maintenance(arch, traffic_light, tag):\n", + " \"\"\"\n", + " Perform maintenance on traffic light\n", + " \"\"\"\n", + " maintenance_performed = arch.events.create(\n", + " traffic_light[\"identity\"],\n", + " {\n", + " \"operation\": \"Record\",\n", + " \"behaviour\": \"RecordEvidence\",\n", + " },\n", + " {\n", + " \"arc_description\": \"Maintenance performed on traffic light\",\n", + " \"arc_display_type\": f\"Maintenance Performed {tag}\",\n", + " },\n", + " confirm=True,\n", + " )\n", + " print(\"MAINTENANCE_PERFORMED:\", json_dumps(maintenance_performed, indent=4))" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "ec1909a6", + "metadata": {}, + "outputs": [], + "source": [ + "\"\"\"Main function of create event.\n", + "\n", + "Parse in user input of url and auth token and use them to\n", + "create an example archivist connection and create an asset.\n", + "\n", + "\"\"\"\n", + "# optional call to set the logger level for all subsystems. The argument can\n", + "# be either \"INFO\" or \"DEBUG\". For more sophisticated logging control see the\n", + "# documentation.\n", + "set_logger(\"INFO\")\n", + "\n", + "# Get authorization token. The token grants certain rights and access permissions.\n", + "# The token can represent the root principal or user in an organization. Different tokens\n", + "# could indicate different users in the same organization or membership of different\n", + "# organisations.\n", + "\n", + "with open(\"/home/builder/credentials/token\", mode=\"r\", encoding=\"utf-8\") as tokenfile:\n", + " auth = tokenfile.read().strip()\n", + "\n", + "# Initialize connection to Archivist\n", + "arch = Archivist(\n", + " URL,\n", + " auth,\n", + " max_time=300,\n", + " verify=False,\n", + ")" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "b7482420", + "metadata": { + "scrolled": true + }, + "outputs": [], + "source": [ + "tag = uuid4() # make this example repeatable" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "bb5b0651", + "metadata": {}, + "outputs": [], + "source": [ + "# make a SINCE compliance policy that alerts when the\n", + "# maintenance performed event has expired.\n", + "compliance_policy = create_compliance_policy(arch, tag)" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "f8ac6dda", + "metadata": {}, + "outputs": [], + "source": [ + "# create an asset that matches the assets_filter field in the\n", + "# compliance policy.\n", + "traffic_light = create_traffic_light(arch)" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "d93be01f", + "metadata": {}, + "outputs": [], + "source": [ + "# perform maintenance on the asset which is valid for 10 seconds.\n", + "perform_maintenance(arch, traffic_light, tag)\n", + "\n", + "# and check compliance - should be OK.\n", + "print(\"Sleep 1 second...\")\n", + "sleep(1)\n", + "compliance = arch.compliance.compliant_at(\n", + " traffic_light[\"identity\"],\n", + ")\n", + "print(\"COMPLIANCE (true):\", json_dumps(compliance, indent=4))\n", + "\n", + "# however waiting long enough (> 10s) will cause the asset to\n", + "# become non-compliant...\n", + "print(\"Sleep 15 seconds...\")\n", + "sleep(15)\n", + "compliance = arch.compliance.compliant_at(\n", + " traffic_light[\"identity\"],\n", + ")\n", + "print(\"COMPLIANCE (false):\", json_dumps(compliance, indent=4))" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "40ffc716", + "metadata": {}, + "outputs": [], + "source": [ + "# finally delete the compliance_policy\n", + "_ = arch.compliance_policies.delete(\n", + " compliance_policy[\"identity\"],\n", + ")" + ] + } + ], + "metadata": { + "kernelspec": { + "display_name": "Python 3 (ipykernel)", + "language": "python", + "name": "python3" + }, + "language_info": { + "codemirror_mode": { + "name": "ipython", + "version": 3 + }, + "file_extension": ".py", + "mimetype": "text/x-python", + "name": "python", + "nbconvert_exporter": "python", + "pygments_lexer": "ipython3", + "version": "3.7.13" + } + }, + "nbformat": 4, + "nbformat_minor": 5 +} diff --git a/unittests/testcompliance_policy_request.py b/unittests/testcompliance_policy_request.py index d06ee941..273b824a 100644 --- a/unittests/testcompliance_policy_request.py +++ b/unittests/testcompliance_policy_request.py @@ -57,6 +57,7 @@ def test_compliance_policy_current_outstanding(self): ["x", "z"], ], event_display_type="current_outstanding event_display_type", + closing_event_display_type="current_outstanding closing_event_display_type", ).dict(), { "compliance_type": "COMPLIANCE_CURRENT_OUTSTANDING", @@ -67,6 +68,7 @@ def test_compliance_policy_current_outstanding(self): {"or": ["x", "z"]}, ], "event_display_type": "current_outstanding event_display_type", + "closing_event_display_type": "current_outstanding closing_event_display_type", }, msg="Incorrect dictionary", ) @@ -115,7 +117,7 @@ def test_compliance_policy_dynamic_tolerance(self): ["x", "z"], ], event_display_type="dynamic_tolerance event_display_type", - closing_event_display_type="period_outstanding closing_event_display_type", + closing_event_display_type="dynamic_tolerance closing_event_display_type", dynamic_window=86400, dynamic_variability=0.5, ).dict(), @@ -128,7 +130,7 @@ def test_compliance_policy_dynamic_tolerance(self): {"or": ["x", "z"]}, ], "event_display_type": "dynamic_tolerance event_display_type", - "closing_event_display_type": "period_outstanding closing_event_display_type", + "closing_event_display_type": "dynamic_tolerance closing_event_display_type", "dynamic_window": 86400, "dynamic_variability": 0.5, },