From 75f9479c6efae03e43e73571ee2ed68b0d0e04aa Mon Sep 17 00:00:00 2001 From: Tomos Williams Date: Tue, 1 Oct 2024 16:17:24 +0100 Subject: [PATCH 1/3] removed special character validation --- api/applications/views/tests/test_parties.py | 79 -------------------- api/core/tests/test_validators.py | 20 ----- api/core/validators.py | 25 ------- api/goods/serializers.py | 11 ++- api/goods/tests/test_serializers.py | 48 ------------ api/parties/serializers.py | 5 +- api/parties/tests/test_serializers.py | 57 -------------- 7 files changed, 7 insertions(+), 238 deletions(-) delete mode 100644 api/core/tests/test_validators.py diff --git a/api/applications/views/tests/test_parties.py b/api/applications/views/tests/test_parties.py index 2492e2f198..68bd1b50cd 100644 --- a/api/applications/views/tests/test_parties.py +++ b/api/applications/views/tests/test_parties.py @@ -84,85 +84,6 @@ def setUp(self): self.party_on_application = PartyOnApplicationFactory(application=self.application) - @parameterized.expand( - [ - ( - {"name": "end_user", "address": "1 Example Street"}, - True, - { - "name": [ - "Party name must only include letters, numbers, and common special characters such as hyphens, brackets and apostrophes" - ] - }, - ), - ( - {"name": "end\auser", "address": "1 Example Street"}, - True, - { - "name": [ - "Party name must only include letters, numbers, and common special characters such as hyphens, brackets and apostrophes" - ] - }, - ), - ( - {"name": "end£user", "address": "1 Example Street"}, - True, - { - "name": [ - "Party name must only include letters, numbers, and common special characters such as hyphens, brackets and apostrophes" - ] - }, - ), - ( - {"name": "end user", "address": "1_Example Street"}, - True, - { - "address": [ - "Address must only include letters, numbers, and common special characters such as hyphens, brackets and apostrophes" - ] - }, - ), - ( - {"name": "end user", "address": "1\aExample Street"}, - True, - { - "address": [ - "Address must only include letters, numbers, and common special characters such as hyphens, brackets and apostrophes" - ] - }, - ), - ( - {"name": "end_user", "address": "1\aExample Street"}, - True, - { - "name": [ - "Party name must only include letters, numbers, and common special characters such as hyphens, brackets and apostrophes" - ], - "address": [ - "Address must only include letters, numbers, and common special characters such as hyphens, brackets and apostrophes" - ], - }, - ), - ] - ) - def test_party_post_invalid(self, data, error, error_message): - data["country"] = {"id": "FR", "name": "France"} - self.url = reverse( - "applications:party", - kwargs={"pk": str(self.application.pk), "party_pk": str(self.party_on_application.party.pk)}, - ) - party = Party.objects.get(id=self.party_on_application.party.pk) - versions = Version.objects.get_for_object(party) - self.assertEqual(versions.count(), 0) - - self.application.status = CaseStatus.objects.get(status="draft") - self.application.save() - - response = self.client.put(self.url, **self.exporter_headers, data=data) - if error: - self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) - self.assertEqual(response.json()["errors"], error_message) - @parameterized.expand( [ ({"name": "end user", "address": "1 Example Street"},), diff --git a/api/core/tests/test_validators.py b/api/core/tests/test_validators.py deleted file mode 100644 index aea29f02d0..0000000000 --- a/api/core/tests/test_validators.py +++ /dev/null @@ -1,20 +0,0 @@ -import pytest -from rest_framework.exceptions import ValidationError -from api.core.validators import EdifactStringValidator - - -@pytest.mark.parametrize( - "value", - (("random value"), ("random-value"), ("random!value"), ("random-!.<>/%&*;+'(),.value")), -) -def test_edifactstringvalidator_valid(value): - validator = EdifactStringValidator() - result = validator(value) - assert result is None - - -@pytest.mark.parametrize("value", (("\r\n"), ("random_value"), ("random$value"), ("random@value"))) -def test_edifactstringvalidator_invalid(value): - validator = EdifactStringValidator() - with pytest.raises(ValidationError): - results = validator(value) diff --git a/api/core/validators.py b/api/core/validators.py index b41459ca75..db4a44dff1 100644 --- a/api/core/validators.py +++ b/api/core/validators.py @@ -1,6 +1,5 @@ from django.utils.deconstruct import deconstructible from rest_framework.exceptions import ValidationError -import re from api.staticdata.control_list_entries.models import ControlListEntry @@ -22,27 +21,3 @@ def __call__(self, value): ControlListEntry.objects.get(rating=value) except ControlListEntry.DoesNotExist: raise ValidationError(self.message, code=self.code) - - -class EdifactStringValidator: - message = "Undefined Error" - regex_string = r"^[a-zA-Z0-9 .,\-\)\(\/'+:=\?\!\"%&\*;\<\>]+$" - - def __call__(self, value): - match_regex = re.compile(self.regex_string) - is_value_valid = bool(match_regex.match(value)) - if not is_value_valid: - raise ValidationError(self.message) - - -class GoodNameValidator(EdifactStringValidator): - message = "Product name must only include letters, numbers, and common special characters such as hyphens, brackets and apostrophes" - - -class PartyAddressValidator(EdifactStringValidator): - regex_string = re.compile(r"^[a-zA-Z0-9 .,\-\)\(\/'+:=\?\!\"%&\*;\<\>\r\n]+$") - message = "Address must only include letters, numbers, and common special characters such as hyphens, brackets and apostrophes" - - -class PartyNameValidator(EdifactStringValidator): - message = "Party name must only include letters, numbers, and common special characters such as hyphens, brackets and apostrophes" diff --git a/api/goods/serializers.py b/api/goods/serializers.py index 122c66e779..379f7036c0 100644 --- a/api/goods/serializers.py +++ b/api/goods/serializers.py @@ -2,7 +2,6 @@ from rest_framework.relations import PrimaryKeyRelatedField from api.core.helpers import str_to_bool from api.core.serializers import KeyValueChoiceField, ControlListEntryField, GoodControlReviewSerializer -from api.core.validators import GoodNameValidator from api.documents.libraries.process_document import process_document from api.goods.enums import ( FirearmCategory, @@ -308,7 +307,7 @@ def update(self, instance, validated_data): class GoodListSerializer(serializers.Serializer): id = serializers.UUIDField() - name = serializers.CharField(validators=[GoodNameValidator()]) + name = serializers.CharField() description = serializers.CharField() control_list_entries = ControlListEntrySerializer(many=True, allow_null=True) part_number = serializers.CharField() @@ -357,7 +356,7 @@ class GoodCreateSerializer(serializers.ModelSerializer): Because of this, each 'get' override must check the instance type before creating queries """ - name = serializers.CharField(error_messages={"blank": "Enter a product name"}, validators=[GoodNameValidator()]) + name = serializers.CharField(error_messages={"blank": "Enter a product name"}) description = serializers.CharField(max_length=280, allow_blank=True, required=False) is_good_controlled = KeyValueChoiceField(choices=GoodControlled.choices, allow_null=True) control_list_entries = ControlListEntryField(required=False, many=True, allow_null=True, allow_empty=True) @@ -690,7 +689,7 @@ def create(self, validated_data): class GoodDocumentViewSerializer(serializers.Serializer): id = serializers.UUIDField() created_at = serializers.DateTimeField() - name = serializers.CharField(validators=[GoodNameValidator()]) + name = serializers.CharField() description = serializers.CharField() user = ExporterUserSimpleSerializer() s3_key = serializers.SerializerMethodField() @@ -788,7 +787,7 @@ class Meta: class GoodSerializerInternal(serializers.Serializer): id = serializers.UUIDField() - name = serializers.CharField(validators=[GoodNameValidator()]) + name = serializers.CharField() description = serializers.CharField() part_number = serializers.CharField() no_part_number_comments = serializers.CharField() @@ -871,7 +870,7 @@ def get_user(self, instance): class GoodSerializerExporter(serializers.Serializer): id = serializers.UUIDField() - name = serializers.CharField(validators=[GoodNameValidator()]) + name = serializers.CharField() description = serializers.CharField() control_list_entries = ControlListEntryField(many=True) part_number = serializers.CharField() diff --git a/api/goods/tests/test_serializers.py b/api/goods/tests/test_serializers.py index 6f17034ab9..fd43cf6e31 100644 --- a/api/goods/tests/test_serializers.py +++ b/api/goods/tests/test_serializers.py @@ -170,30 +170,6 @@ def test_validate_good_internal_name_valid(self, name): [ ("", "This field may not be blank."), ("\r\n", "This field may not be blank."), - ( - "good\rname", - "Product name must only include letters, numbers, and common special characters such as hyphens, brackets and apostrophes", - ), - ( - "good\nname", - "Product name must only include letters, numbers, and common special characters such as hyphens, brackets and apostrophes", - ), - ( - "good\r\nname", - "Product name must only include letters, numbers, and common special characters such as hyphens, brackets and apostrophes", - ), - ( - "good_name", - "Product name must only include letters, numbers, and common special characters such as hyphens, brackets and apostrophes", - ), - ( - "good$name", - "Product name must only include letters, numbers, and common special characters such as hyphens, brackets and apostrophes", - ), - ( - "good@name", - "Product name must only include letters, numbers, and common special characters such as hyphens, brackets and apostrophes", - ), ] ) def test_validate_good_internal_name_invalid(self, name, error_message): @@ -258,30 +234,6 @@ def test_validate_good_exporter_name_valid(self, address): [ ("", "This field may not be blank."), ("\r\n", "This field may not be blank."), - ( - "good\rname", - "Product name must only include letters, numbers, and common special characters such as hyphens, brackets and apostrophes", - ), - ( - "good\nname", - "Product name must only include letters, numbers, and common special characters such as hyphens, brackets and apostrophes", - ), - ( - "good\r\nname", - "Product name must only include letters, numbers, and common special characters such as hyphens, brackets and apostrophes", - ), - ( - "good_name", - "Product name must only include letters, numbers, and common special characters such as hyphens, brackets and apostrophes", - ), - ( - "good$name", - "Product name must only include letters, numbers, and common special characters such as hyphens, brackets and apostrophes", - ), - ( - "good@name", - "Product name must only include letters, numbers, and common special characters such as hyphens, brackets and apostrophes", - ), ] ) def test_validate_good_exporter_name_invalid(self, name, error_message): diff --git a/api/parties/serializers.py b/api/parties/serializers.py index 58a5eb850e..f51e89bec0 100644 --- a/api/parties/serializers.py +++ b/api/parties/serializers.py @@ -3,7 +3,6 @@ from api.cases.enums import CaseTypeSubTypeEnum from api.core.serializers import KeyValueChoiceField, CountrySerializerField -from api.core.validators import PartyAddressValidator, PartyNameValidator from api.documents.libraries.process_document import process_document from api.flags.serializers import FlagSerializer from api.goods.enums import PvGrading @@ -15,8 +14,8 @@ class PartySerializer(serializers.ModelSerializer): - name = serializers.CharField(error_messages=PartyErrors.NAME, validators=[PartyNameValidator()]) - address = serializers.CharField(error_messages=PartyErrors.ADDRESS, validators=[PartyAddressValidator()]) + name = serializers.CharField(error_messages=PartyErrors.NAME) + address = serializers.CharField(error_messages=PartyErrors.ADDRESS) country = CountrySerializerField() website = serializers.CharField(required=False, allow_blank=True) signatory_name_euu = serializers.CharField(allow_blank=True) diff --git a/api/parties/tests/test_serializers.py b/api/parties/tests/test_serializers.py index 2d2b86a9d1..4d6e5e9c50 100644 --- a/api/parties/tests/test_serializers.py +++ b/api/parties/tests/test_serializers.py @@ -58,26 +58,6 @@ def test_validate_party_address_valid(self, address): @parameterized.expand( [ ("\r\n", "Enter an address"), - ( - "party\address", - "Address must only include letters, numbers, and common special characters such as hyphens, brackets and apostrophes", - ), - ( - "party-\waddress", - "Address must only include letters, numbers, and common special characters such as hyphens, brackets and apostrophes", - ), - ( - "party_address", - "Address must only include letters, numbers, and common special characters such as hyphens, brackets and apostrophes", - ), - ( - "party$address", - "Address must only include letters, numbers, and common special characters such as hyphens, brackets and apostrophes", - ), - ( - "party@address", - "Address must only include letters, numbers, and common special characters such as hyphens, brackets and apostrophes", - ), ] ) def test_validate_party_address_invalid(self, address, error_message): @@ -107,40 +87,3 @@ def test_validate_party_name_valid(self, name): partial=True, ) self.assertTrue(serializer.is_valid()) - - @parameterized.expand( - [ - ( - "party\aname", - "Party name must only include letters, numbers, and common special characters such as hyphens, brackets and apostrophes", - ), - ( - "party-\wname", - "Party name must only include letters, numbers, and common special characters such as hyphens, brackets and apostrophes", - ), - ( - "party_name", - "Party name must only include letters, numbers, and common special characters such as hyphens, brackets and apostrophes", - ), - ( - "party$name", - "Party name must only include letters, numbers, and common special characters such as hyphens, brackets and apostrophes", - ), - ( - "party@name", - "Party name must only include letters, numbers, and common special characters such as hyphens, brackets and apostrophes", - ), - ] - ) - def test_party_name_invalid(self, name, error_message): - serializer = PartySerializer( - data={"name": name}, - partial=True, - ) - self.assertFalse(serializer.is_valid()) - serializer_error = serializer.errors["name"] - self.assertEqual(len(serializer_error), 1) - self.assertEqual( - str(serializer_error[0]), - error_message, - ) From aafbf09b4eadd27074bb2f72a7ddd95571a616f5 Mon Sep 17 00:00:00 2001 From: Kevin Carrogan Date: Wed, 9 Oct 2024 12:29:48 +0100 Subject: [PATCH 2/3] Move codeowners file to correct folder --- {.circleci/.github => .github}/CODEOWNERS | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename {.circleci/.github => .github}/CODEOWNERS (100%) diff --git a/.circleci/.github/CODEOWNERS b/.github/CODEOWNERS similarity index 100% rename from .circleci/.github/CODEOWNERS rename to .github/CODEOWNERS From f4717046a5510745ae76e3f21bf750f0f60ce3a2 Mon Sep 17 00:00:00 2001 From: Kevin Carrogan Date: Tue, 1 Oct 2024 12:30:48 +0100 Subject: [PATCH 3/3] Move flagging rule functions into lite_routing --- api/applications/views/applications.py | 2 +- api/cases/signals.py | 2 +- api/flags/views.py | 5 +- api/organisations/views/organisations.py | 2 +- api/queries/end_user_advisories/views.py | 2 +- api/queries/goods_query/views.py | 2 +- api/workflow/flagging_rules_automation.py | 270 ------------------ api/workflow/tests/test_flagging_rules.py | 320 ---------------------- lite_routing | 2 +- test_helpers/clients.py | 2 +- 10 files changed, 11 insertions(+), 598 deletions(-) delete mode 100644 api/workflow/flagging_rules_automation.py delete mode 100644 api/workflow/tests/test_flagging_rules.py diff --git a/api/applications/views/applications.py b/api/applications/views/applications.py index 7af74cba30..b9696b7761 100644 --- a/api/applications/views/applications.py +++ b/api/applications/views/applications.py @@ -97,7 +97,7 @@ from api.staticdata.statuses.models import CaseSubStatus from api.users.libraries.notifications import get_case_notifications from api.users.models import ExporterUser -from api.workflow.flagging_rules_automation import apply_flagging_rules_to_case +from lite_routing.routing_rules_internal.flagging_engine import apply_flagging_rules_to_case from lite_routing.routing_rules_internal.routing_engine import run_routing_rules diff --git a/api/cases/signals.py b/api/cases/signals.py index 05cf39228d..2e73e31a6f 100644 --- a/api/cases/signals.py +++ b/api/cases/signals.py @@ -5,7 +5,7 @@ from api.cases.models import Case from api.staticdata.statuses.enums import CaseStatusEnum from api.staticdata.statuses.libraries.get_case_status import get_case_status_by_status -from api.workflow.flagging_rules_automation import apply_flagging_rules_to_case +from lite_routing.routing_rules_internal.flagging_engine import apply_flagging_rules_to_case @receiver(pre_save) diff --git a/api/flags/views.py b/api/flags/views.py index 4fc3e8950f..590fcf9e37 100644 --- a/api/flags/views.py +++ b/api/flags/views.py @@ -43,7 +43,10 @@ from api.queries.end_user_advisories.models import EndUserAdvisoryQuery from api.queries.goods_query.models import GoodsQuery -from api.workflow.flagging_rules_automation import apply_flagging_rule_to_all_open_cases, apply_flagging_rule_for_flag +from lite_routing.routing_rules_internal.flagging_engine import ( + apply_flagging_rule_to_all_open_cases, + apply_flagging_rule_for_flag, +) from lite_content.lite_api import strings diff --git a/api/organisations/views/organisations.py b/api/organisations/views/organisations.py index 163c6a8213..212bb92c06 100644 --- a/api/organisations/views/organisations.py +++ b/api/organisations/views/organisations.py @@ -37,7 +37,7 @@ from api.staticdata.statuses.libraries.get_case_status import get_case_status_by_status from api.staticdata.statuses.models import CaseStatus from api.users.enums import UserType -from api.workflow.flagging_rules_automation import apply_flagging_rules_to_case +from lite_routing.routing_rules_internal.flagging_engine import apply_flagging_rules_to_case class OrganisationsList(generics.ListCreateAPIView): diff --git a/api/queries/end_user_advisories/views.py b/api/queries/end_user_advisories/views.py index 0e4ddd87e3..f4b99fce60 100644 --- a/api/queries/end_user_advisories/views.py +++ b/api/queries/end_user_advisories/views.py @@ -13,7 +13,7 @@ from api.queries.end_user_advisories.models import EndUserAdvisoryQuery from api.queries.end_user_advisories.serializers import EndUserAdvisoryViewSerializer, EndUserAdvisoryListSerializer from api.users.libraries.notifications import get_case_notifications -from api.workflow.flagging_rules_automation import apply_flagging_rules_to_case +from lite_routing.routing_rules_internal.flagging_engine import apply_flagging_rules_to_case class EndUserAdvisoriesList(ListAPIView): diff --git a/api/queries/goods_query/views.py b/api/queries/goods_query/views.py index ba24cd9199..87e9e2b039 100644 --- a/api/queries/goods_query/views.py +++ b/api/queries/goods_query/views.py @@ -26,7 +26,7 @@ from api.queries.helpers import get_exporter_query from api.staticdata.statuses.enums import CaseStatusEnum from api.users.models import UserOrganisationRelationship -from api.workflow.flagging_rules_automation import apply_flagging_rules_to_case +from lite_routing.routing_rules_internal.flagging_engine import apply_flagging_rules_to_case class GoodsQueriesCreate(APIView): diff --git a/api/workflow/flagging_rules_automation.py b/api/workflow/flagging_rules_automation.py deleted file mode 100644 index 3cdd77c885..0000000000 --- a/api/workflow/flagging_rules_automation.py +++ /dev/null @@ -1,270 +0,0 @@ -from django.db.models import Q, QuerySet - -from api.applications.models import PartyOnApplication, GoodOnApplication -from api.cases.enums import CaseTypeEnum -from api.cases.models import Case -from api.flags.enums import FlagLevels, FlagStatuses -from api.flags.models import FlaggingRule, Flag -from api.goods.enums import GoodStatus -from api.goods.models import Good -from api.goodstype.models import GoodsType -from api.parties.models import Party -from api.queries.end_user_advisories.models import EndUserAdvisoryQuery -from api.queries.goods_query.models import GoodsQuery -from api.staticdata.countries.models import Country -from api.staticdata.statuses.enums import CaseStatusEnum -from api.staticdata.control_list_entries.helpers import get_clc_child_nodes, get_clc_parent_nodes - -from lite_routing.routing_rules_internal.flagging_rules_criteria import ( - run_flagging_rules_criteria_case, - run_flagging_rules_criteria_product, - run_flagging_rules_criteria_destination, -) - - -def _apply_python_flagging_rules(level, object): - flags = [] - - for rule in FlaggingRule.objects.filter(level=level, status=FlagStatuses.ACTIVE, is_python_criteria=True): - rule_applies = False - if level == FlagLevels.CASE: - rule_applies = run_flagging_rules_criteria_case(rule.id, object) - elif level == FlagLevels.GOOD: - rule_applies = run_flagging_rules_criteria_product(rule.id, object) - elif level == FlagLevels.DESTINATION: - rule_applies = run_flagging_rules_criteria_destination(rule.id, object) - - if rule_applies: - flags.append(rule.flag_id) - - return flags - - -def get_active_legacy_flagging_rules_for_level(level): - return FlaggingRule.objects.prefetch_related("flag").filter( - status=FlagStatuses.ACTIVE, flag__status=FlagStatuses.ACTIVE, level=level, is_python_criteria=False - ) - - -def apply_flagging_rules_to_case(case): - """ - Apply all active flagging rules to a case which meet the criteria - """ - # flagging rules should only be applied to cases which are open - if case.status.status == CaseStatusEnum.DRAFT or CaseStatusEnum.is_terminal(case.status.status): - return - - apply_case_flagging_rules(case) - apply_destination_flagging_rules_for_case(case) - apply_good_flagging_rules_for_case(case) - - -def apply_case_flagging_rules(case): - """ - Applies case type flagging rules to a case object - """ - # get a list of flag_id's where the flagging rule matching value is equivalent to the case_type - flags = ( - get_active_legacy_flagging_rules_for_level(FlagLevels.CASE) - .filter(matching_values__overlap=[case.case_type.reference]) - .values_list("flag_id", flat=True) - ) - - flags = list(flags) - flags.extend(_apply_python_flagging_rules(FlagLevels.CASE, case)) - - if flags: - case.flags.add(*flags) - - -def apply_destination_flagging_rules_for_case(case, flagging_rule: QuerySet = None): - """ - Applies destination type flagging rules to a case object - """ - # If the flagging rules are specified then these is the only one we expect, else get all active - flagging_rules = ( - get_active_legacy_flagging_rules_for_level(FlagLevels.DESTINATION) if not flagging_rule else flagging_rule - ) - - if case.case_type.id == CaseTypeEnum.EUA.id: - # if the instance is not an EndUserAdvisoryQuery we need to get this to have access to the party - if not isinstance(case, EndUserAdvisoryQuery): - case = EndUserAdvisoryQuery.objects.get(pk=case.id) - parties = [case.end_user] - else: - parties = Party.objects.filter( - id__in=PartyOnApplication.objects.filter(application_id=case.id).values("party_id") - ) - - for party in parties: - apply_destination_rule_on_party(party, flagging_rules) - - -def apply_destination_rule_on_party(party: Party, flagging_rules: QuerySet = None): - # If the flagging rules are specified then these is the only one we expect, else get all active - flagging_rules = ( - get_active_legacy_flagging_rules_for_level(FlagLevels.DESTINATION) if not flagging_rules else flagging_rules - ) - - # get a list of flag_id's where the flagging rule matching value is equivalent to the country id - flags = flagging_rules.filter(matching_values__overlap=[party.country.id]).values_list("flag_id", flat=True) - - flags = list(flags) - flags.extend(_apply_python_flagging_rules(FlagLevels.DESTINATION, party)) - - if flags: - party.flags.add(*flags) - - -def apply_good_flagging_rules_for_case(case, flagging_rule: QuerySet = None): - # If the flagging rules are specified then these is the only one we expect, else get all active - flagging_rules = get_active_legacy_flagging_rules_for_level(FlagLevels.GOOD) if not flagging_rule else flagging_rule - - if case.case_type.id == CaseTypeEnum.GOODS.id: - if not isinstance(case, GoodsQuery): - case = GoodsQuery.objects.get(pk=case.id) - goods = [case.good] - elif case.case_type_id in [CaseTypeEnum.OICL.id, CaseTypeEnum.OGEL.id, CaseTypeEnum.OIEL.id, CaseTypeEnum.HMRC.id]: - goods = GoodsType.objects.filter(application_id=case.id) - else: - goods = ( - Good.objects.prefetch_related("goods_on_application") - .filter(goods_on_application__application_id=case.id) - .distinct() - ) - - for good in goods: - apply_goods_rules_for_good(good, flagging_rules) - - -def apply_goods_rules_for_good(good, flagging_rules: QuerySet = None): - # If the flagging rules are specified then these is the only one we expect, else get all active - flagging_rules = ( - get_active_legacy_flagging_rules_for_level(FlagLevels.GOOD) if not flagging_rules else flagging_rules - ) - - # get a list of flag_id's where the flagging rule matching value is equivalent to the good control code - ratings = [r for r in good.control_list_entries.values_list("rating", flat=True)] - group_ratings = [] - for rating in ratings: - group_ratings.extend(get_clc_parent_nodes(rating)) - - flagging_rules = flagging_rules.filter( - Q(matching_values__overlap=ratings) | Q(matching_groups__overlap=group_ratings) - ).exclude(excluded_values__overlap=(ratings + group_ratings)) - - if isinstance(good, Good) and good.status != GoodStatus.VERIFIED: - flagging_rules = flagging_rules.exclude(is_for_verified_goods_only=True) - - flags = flagging_rules.values_list("flag_id", flat=True) - - flags = list(flags) - flags.extend(_apply_python_flagging_rules(FlagLevels.GOOD, good)) - - if flags: - good.flags.add(*flags) - - -def apply_flagging_rule_to_all_open_cases(flagging_rule: FlaggingRule): - """ - Takes a flagging rule and creates a relationship between it's flag and objects that meet match conditions - """ - if flagging_rule.status == FlagStatuses.ACTIVE and flagging_rule.flag.status == FlagStatuses.ACTIVE: - # Flagging rules should only be applied to open cases - draft_and_terminal_statuses = [CaseStatusEnum.DRAFT, *CaseStatusEnum.terminal_statuses()] - open_cases = Case.objects.exclude(status__status__in=draft_and_terminal_statuses) - - # Apply the flagging rule to different entities depending on the rule's level - if flagging_rule.level == FlagLevels.CASE: - # Add flag to all open Cases - open_cases = open_cases.filter(case_type__reference__in=flagging_rule.matching_values).values_list( - "id", flat=True - ) - - flagging_rule.flag.cases.add(*open_cases) - - elif flagging_rule.level == FlagLevels.GOOD: - clc_entries_of_groups = [] - for group in flagging_rule.matching_groups: - child_entries = get_clc_child_nodes(group) - clc_entries_of_groups.extend(child_entries) - - matching_values = flagging_rule.matching_values + clc_entries_of_groups - - # excluded_values contain individual entries and groups - excluded_values = [] - for rating in flagging_rule.excluded_values: - entries = get_clc_child_nodes(rating) - excluded_values.extend(entries) - - # Add flag to all Goods on open Goods Queries - goods_in_query = GoodsQuery.objects.filter(good__control_list_entries__rating__in=matching_values).exclude( - status__status__in=draft_and_terminal_statuses - ) - - if excluded_values: - # exclusion entries - goods that doesn't contain given control list entries - goods_in_query = goods_in_query.exclude(good__control_list_entries__rating__in=excluded_values) - - if flagging_rule.is_for_verified_goods_only: - goods_in_query = goods_in_query.filter(good__status=GoodStatus.VERIFIED) - - goods_in_query = goods_in_query.values_list("good_id", flat=True) - flagging_rule.flag.goods.add(*goods_in_query) - - # Add flag to all Goods Types - goods_types = GoodsType.objects.filter( - application_id__in=open_cases, control_list_entries__rating__in=matching_values - ) - - if excluded_values: - goods_types = goods_types.exclude( - application_id__in=open_cases, control_list_entries__rating__in=excluded_values - ) - - goods_types = goods_types.values_list("id", flat=True) - - flagging_rule.flag.goods_type.add(*goods_types) - - # Add flag to all open Applications - goods = GoodOnApplication.objects.filter( - application_id__in=open_cases, good__control_list_entries__rating__in=matching_values - ) - - if excluded_values: - goods = goods.exclude( - application_id__in=open_cases, good__control_list_entries__rating__in=excluded_values - ) - - if flagging_rule.is_for_verified_goods_only: - goods = goods.filter(good__status=GoodStatus.VERIFIED) - - goods = goods.values_list("good_id", flat=True) - flagging_rule.flag.goods.add(*goods) - - elif flagging_rule.level == FlagLevels.DESTINATION: - # Add flag to all End Users on open End User Advisory Queries - end_users = ( - EndUserAdvisoryQuery.objects.filter(end_user__country_id__in=flagging_rule.matching_values) - .exclude(status__status__in=draft_and_terminal_statuses) - .values_list("end_user_id", flat=True) - ) - flagging_rule.flag.parties.add(*end_users) - - # Add flag to all Parties on open Applications - parties = PartyOnApplication.objects.filter( - application_id__in=open_cases, party__country_id__in=flagging_rule.matching_values - ).values_list("party_id", flat=True) - flagging_rule.flag.parties.add(*parties) - - countries = Country.objects.filter(id__in=flagging_rule.matching_values).values_list("id", flat=True) - flagging_rule.flag.countries.add(*countries) - - -def apply_flagging_rule_for_flag(flag: Flag): - """ - gets the flagging rules relating to a flag and applies them - """ - flagging_rules = FlaggingRule.objects.filter(flag=flag) - for rule in flagging_rules: - apply_flagging_rule_to_all_open_cases(rule) diff --git a/api/workflow/tests/test_flagging_rules.py b/api/workflow/tests/test_flagging_rules.py deleted file mode 100644 index 3607e359fb..0000000000 --- a/api/workflow/tests/test_flagging_rules.py +++ /dev/null @@ -1,320 +0,0 @@ -import unittest - -from parameterized import parameterized - -from api.applications.models import GoodOnApplication, PartyOnApplication -from api.applications.tests.factories import PartyOnApplicationFactory -from api.flags.enums import FlagLevels, FlagStatuses -from api.flags.models import Flag, FlaggingRule -from api.goods.enums import GoodStatus -from api.goods.tests.factories import GoodFactory -from api.staticdata.control_list_entries.helpers import get_control_list_entry -from api.staticdata.statuses.enums import CaseStatusEnum -from api.staticdata.statuses.libraries.get_case_status import get_case_status_by_status -from api.teams.models import Team -from test_helpers.clients import DataTestClient -from api.workflow.flagging_rules_automation import ( - apply_flagging_rules_to_case, - apply_case_flagging_rules, - apply_goods_rules_for_good, - apply_destination_rule_on_party, - get_active_legacy_flagging_rules_for_level, - apply_flagging_rule_to_all_open_cases, -) - - -class FlaggingRulesAutomation(DataTestClient): - def test_get_active_flagging_rules_goods(self): - active_flag = self.create_flag(name="good flag", level=FlagLevels.GOOD, team=self.team) - self.create_flagging_rule(level=FlagLevels.GOOD, team=self.team, flag=active_flag, matching_values=["abc"]) - - deactivated_flag = self.create_flag(name="good flag 2", level=FlagLevels.GOOD, team=self.team) - self.create_flagging_rule( - level=FlagLevels.GOOD, - team=self.team, - flag=deactivated_flag, - matching_values=["abc"], - status=FlagStatuses.DEACTIVATED, - ) - - flagging_rules = list(get_active_legacy_flagging_rules_for_level(level=FlagLevels.GOOD)) - - self.assertTrue(active_flag in [rule.flag for rule in flagging_rules]) - self.assertTrue(deactivated_flag not in [rule.flag for rule in flagging_rules]) - - def test_get_active_flag_flagging_rules_goods(self): - active_flag = self.create_flag(name="good flag", level=FlagLevels.GOOD, team=self.team) - self.create_flagging_rule(level=FlagLevels.GOOD, team=self.team, flag=active_flag, matching_values=["abc"]) - - deactivated_flag = self.create_flag(name="good flag 2", level=FlagLevels.GOOD, team=self.team) - self.create_flagging_rule( - level=FlagLevels.GOOD, - team=self.team, - flag=deactivated_flag, - matching_values=["abc"], - ) - deactivated_flag.status = FlagStatuses.DEACTIVATED - deactivated_flag.save() - - flagging_rules = list(get_active_legacy_flagging_rules_for_level(level=FlagLevels.GOOD)) - - self.assertTrue(active_flag in [rule.flag for rule in flagging_rules]) - self.assertTrue(deactivated_flag not in [rule.flag for rule in flagging_rules]) - - def test_get_active_flagging_rules_destination(self): - active_flag = self.create_flag(name="good flag", level=FlagLevels.DESTINATION, team=self.team) - self.create_flagging_rule( - level=FlagLevels.DESTINATION, team=self.team, flag=active_flag, matching_values=["abc"] - ) - - deactivated_flag = self.create_flag(name="good flag 2", level=FlagLevels.DESTINATION, team=self.team) - self.create_flagging_rule( - level=FlagLevels.DESTINATION, - team=self.team, - flag=deactivated_flag, - matching_values=["abc"], - status=FlagStatuses.DEACTIVATED, - ) - - flagging_rules = list(get_active_legacy_flagging_rules_for_level(level=FlagLevels.DESTINATION)) - - self.assertTrue(active_flag in [rule.flag for rule in flagging_rules]) - self.assertTrue(deactivated_flag not in [rule.flag for rule in flagging_rules]) - - def test_get_active_flag_flagging_rules_destination(self): - active_flag = self.create_flag(name="good flag", level=FlagLevels.DESTINATION, team=self.team) - self.create_flagging_rule( - level=FlagLevels.DESTINATION, team=self.team, flag=active_flag, matching_values=["abc"] - ) - - deactivated_flag = self.create_flag(name="good flag 2", level=FlagLevels.DESTINATION, team=self.team) - self.create_flagging_rule( - level=FlagLevels.DESTINATION, - team=self.team, - flag=deactivated_flag, - matching_values=["abc"], - ) - deactivated_flag.status = FlagStatuses.DEACTIVATED - deactivated_flag.save() - - flagging_rules = list(get_active_legacy_flagging_rules_for_level(level=FlagLevels.DESTINATION)) - - self.assertTrue(active_flag in [rule.flag for rule in flagging_rules]) - self.assertTrue(deactivated_flag not in [rule.flag for rule in flagging_rules]) - - def test_get_active_flagging_rules_case(self): - active_flag = self.create_flag(name="good flag", level=FlagLevels.CASE, team=self.team) - self.create_flagging_rule(level=FlagLevels.CASE, team=self.team, flag=active_flag, matching_values=["abc"]) - - deactivated_flag = self.create_flag(name="good flag 2", level=FlagLevels.CASE, team=self.team) - self.create_flagging_rule( - level=FlagLevels.CASE, - team=self.team, - flag=deactivated_flag, - matching_values=["abc"], - status=FlagStatuses.DEACTIVATED, - ) - - flagging_rules = list(get_active_legacy_flagging_rules_for_level(level=FlagLevels.CASE)) - - self.assertTrue(active_flag in [rule.flag for rule in flagging_rules]) - self.assertTrue(deactivated_flag not in [rule.flag for rule in flagging_rules]) - - def test_get_active_flag_flagging_rules_case(self): - active_flag = self.create_flag(name="good flag", level=FlagLevels.CASE, team=self.team) - self.create_flagging_rule(level=FlagLevels.CASE, team=self.team, flag=active_flag, matching_values=["abc"]) - - deactivated_flag = self.create_flag(name="good flag 2", level=FlagLevels.CASE, team=self.team) - self.create_flagging_rule( - level=FlagLevels.CASE, - team=self.team, - flag=deactivated_flag, - matching_values=["abc"], - ) - deactivated_flag.status = FlagStatuses.DEACTIVATED - deactivated_flag.save() - - flagging_rules = list(get_active_legacy_flagging_rules_for_level(level=FlagLevels.CASE)) - - self.assertTrue(active_flag in [rule.flag for rule in flagging_rules]) - self.assertTrue(deactivated_flag not in [rule.flag for rule in flagging_rules]) - - @parameterized.expand([k for k, v in CaseStatusEnum.get_choices()]) - def test_apply_flagging_rule_to_open_cases(self, case_status): - if case_status == CaseStatusEnum.DRAFT: - case = self.create_draft_standard_application(self.organisation) - else: - case = self.create_standard_application_case(self.organisation) - case.status = get_case_status_by_status(case_status) - case.save() - - flag = self.create_flag(case.case_type.reference, FlagLevels.CASE, self.team) - flagging_rule = self.create_flagging_rule(FlagLevels.CASE, self.team, flag, [case.case_type.reference]) - - apply_flagging_rule_to_all_open_cases(flagging_rule) - - case.refresh_from_db() - - if CaseStatusEnum.is_terminal(case_status) or case_status == CaseStatusEnum.DRAFT: - self.assertNotIn(flag, case.flags.all()) - else: - self.assertIn(flag, case.flags.all()) - - def test_apply_verified_goods_only_flagging_rule_to_open_cases_failure(self): - """Test flag not applied to good when flagging rule is for verified goods only.""" - case = self.create_standard_application_case(self.organisation) - - flag = self.create_flag("good flag", FlagLevels.GOOD, self.team) - flagging_rule = self.create_flagging_rule( - FlagLevels.GOOD, self.team, flag, [case.case_type.reference], is_for_verified_goods_only=True - ) - good = GoodOnApplication.objects.filter(application_id=case.id).first().good - - apply_flagging_rule_to_all_open_cases(flagging_rule) - - case.refresh_from_db() - self.assertNotIn(flag, case.flags.all()) - - def test_apply_verified_goods_only_flagging_rule_to_open_cases_success(self): - case = self.create_standard_application_case(self.organisation) - - flag = self.create_flag("good flag", FlagLevels.GOOD, self.team) - flagging_rule = self.create_flagging_rule( - FlagLevels.GOOD, self.team, flag, [case.case_type.reference], is_for_verified_goods_only=True - ) - good = GoodOnApplication.objects.filter(application_id=case.id).first().good - good.status = GoodStatus.VERIFIED - good.save() - - apply_flagging_rule_to_all_open_cases(flagging_rule) - - case.refresh_from_db() - self.assertNotIn(flag, case.flags.all()) - - @unittest.mock.patch("api.workflow.flagging_rules_automation.run_flagging_rules_criteria_case") - def test_python_criteria_satisfied_calls_case_criteria_function(self, mocked_criteria_function): - mocked_criteria_function.return_value = True - rule = FlaggingRule.objects.create( - team=Team.objects.first(), - flag=Flag.objects.first(), - level=FlagLevels.CASE, - status=FlagStatuses.ACTIVE, - is_python_criteria=True, - ) - case = self.create_standard_application_case(self.organisation) - case.flags.clear() - self.assertEqual(case.flags.count(), 0) - apply_case_flagging_rules(case) - assert mocked_criteria_function.called - self.assertIn(rule.flag, case.flags.all()) - - @unittest.mock.patch("api.workflow.flagging_rules_automation.run_flagging_rules_criteria_product") - def test_python_criteria_satisfied_calls_product_criteria_function(self, mocked_criteria_function): - mocked_criteria_function.return_value = True - rule = FlaggingRule.objects.create( - team=Team.objects.first(), - flag=Flag.objects.first(), - level=FlagLevels.GOOD, - status=FlagStatuses.ACTIVE, - is_python_criteria=True, - ) - good = GoodFactory(organisation=self.organisation) - self.assertEqual(good.flags.count(), 0) - apply_goods_rules_for_good(good) - assert mocked_criteria_function.called - self.assertIn(rule.flag, good.flags.all()) - - @unittest.mock.patch("api.workflow.flagging_rules_automation.run_flagging_rules_criteria_destination") - def test_python_criteria_satisfied_calls_destination_criteria_function(self, mocked_criteria_function): - mocked_criteria_function.return_value = True - rule = FlaggingRule.objects.create( - team=Team.objects.first(), - flag=Flag.objects.first(), - level=FlagLevels.DESTINATION, - status=FlagStatuses.ACTIVE, - is_python_criteria=True, - ) - party_on_application = PartyOnApplicationFactory(party__country__id="US") - party = party_on_application.party - self.assertEqual(party.flags.count(), 0) - apply_destination_rule_on_party(party) - assert mocked_criteria_function.called - self.assertIn(rule.flag, party.flags.all()) - - -class FlaggingRulesAutomationForEachCaseType(DataTestClient): - def test_standard_application(self): - application = self.create_standard_application_case(self.organisation) - - case_flag = self.create_flag("case flag", FlagLevels.CASE, self.team) - self.create_flagging_rule( - FlagLevels.CASE, self.team, flag=case_flag, matching_values=[application.case_type.reference] - ) - - good = GoodOnApplication.objects.filter(application_id=application.id).first().good - good_flag = self.create_flag("good flag", FlagLevels.GOOD, self.team) - self.create_flagging_rule( - FlagLevels.GOOD, self.team, flag=good_flag, matching_values=[good.control_list_entries.first().rating] - ) - - party = PartyOnApplication.objects.filter(application_id=application.id).first().party - destination_flag = self.create_flag("dest flag", FlagLevels.DESTINATION, self.team) - self.create_flagging_rule( - FlagLevels.DESTINATION, self.team, flag=destination_flag, matching_values=[party.country_id] - ) - - self.submit_application(application) - apply_flagging_rules_to_case(application) - - application.refresh_from_db() - good.refresh_from_db() - party.refresh_from_db() - - self.assertIn(case_flag, application.flags.all()) - self.assertIn(good_flag, good.flags.all()) - self.assertIn(destination_flag, party.flags.all()) - - def test_goods_query_application(self): - query = self.create_clc_query("query", self.organisation) - - case_flag = self.create_flag("case flag", FlagLevels.CASE, self.team) - self.create_flagging_rule( - FlagLevels.CASE, self.team, flag=case_flag, matching_values=[query.case_type.reference] - ) - - good = query.good - good.control_list_entries.set([get_control_list_entry("ML1a")]) - good_flag = self.create_flag("good flag", FlagLevels.GOOD, self.team) - self.create_flagging_rule( - FlagLevels.GOOD, self.team, flag=good_flag, matching_values=[good.control_list_entries.first().rating] - ) - - apply_flagging_rules_to_case(query) - - query.refresh_from_db() - good.refresh_from_db() - - self.assertIn(case_flag, query.flags.all()) - self.assertIn(good_flag, good.flags.all()) - - def test_end_user_advisory_application(self): - query = self.create_end_user_advisory("a", "v", self.organisation) - - case_flag = self.create_flag("case flag", FlagLevels.CASE, self.team) - self.create_flagging_rule( - FlagLevels.CASE, self.team, flag=case_flag, matching_values=[query.case_type.reference] - ) - - party = query.end_user - destination_flag = self.create_flag("dest flag", FlagLevels.DESTINATION, self.team) - self.create_flagging_rule( - FlagLevels.DESTINATION, self.team, flag=destination_flag, matching_values=[party.country_id] - ) - - apply_flagging_rules_to_case(query) - - query.refresh_from_db() - party.refresh_from_db() - - self.assertIn(case_flag, query.flags.all()) - self.assertIn(destination_flag, party.flags.all()) diff --git a/lite_routing b/lite_routing index 7cba6774db..7baa85e928 160000 --- a/lite_routing +++ b/lite_routing @@ -1 +1 @@ -Subproject commit 7cba6774db8d6780037bf39a985bee6d6294f245 +Subproject commit 7baa85e9280177c295c64cde72be1aba7ca8e48f diff --git a/test_helpers/clients.py b/test_helpers/clients.py index 933f1f4cb2..ba546025cb 100644 --- a/test_helpers/clients.py +++ b/test_helpers/clients.py @@ -87,7 +87,7 @@ from api.users.enums import SystemUser, UserType from api.users.libraries.user_to_token import user_to_token from api.users.models import ExporterUser, UserOrganisationRelationship, BaseUser, GovUser, Role -from api.workflow.flagging_rules_automation import apply_flagging_rules_to_case +from lite_routing.routing_rules_internal.flagging_engine import apply_flagging_rules_to_case from api.workflow.routing_rules.enum import RoutingRulesAdditionalFields from api.workflow.routing_rules.models import RoutingRule