class Message(models.Model):
    """
    A incoming message from the backend
    """

    TYPE_INBOX = "I"
    TYPE_FLOW = "F"

    TYPE_CHOICES = ((TYPE_INBOX, "Inbox"), (TYPE_FLOW, "Flow"))

    SAVE_CONTACT_ATTR = "__data__contact"
    SAVE_LABELS_ATTR = "__data__labels"

    TIMELINE_TYPE = "I"

    SEARCH_BY_TEXT_DAYS = 90

    org = models.ForeignKey(Org, related_name="incoming_messages", on_delete=models.PROTECT)

    # identifier of the message on the backend
    backend_id = models.IntegerField(unique=True)

    contact = models.ForeignKey(Contact, related_name="incoming_messages", on_delete=models.PROTECT)

    type = models.CharField(max_length=1)

    text = models.TextField(max_length=640)

    labels = models.ManyToManyField(Label, through=Labelling, related_name="messages")

    has_labels = models.BooleanField(default=False)  # maintained via db triggers

    is_flagged = models.BooleanField(default=False)

    is_archived = models.BooleanField(default=False)

    created_on = models.DateTimeField()

    modified_on = models.DateTimeField(null=True, default=now)

    is_handled = models.BooleanField(default=False)

    is_active = models.BooleanField(default=True)

    case = models.ForeignKey("cases.Case", null=True, related_name="incoming_messages", on_delete=models.PROTECT)

    # when this message was last locked by a user
    locked_on = models.DateTimeField(null=True)

    # which user locked it
    locked_by = models.ForeignKey(User, null=True, related_name="actioned_messages", on_delete=models.PROTECT)

    def __init__(self, *args, **kwargs):
        if self.SAVE_CONTACT_ATTR in kwargs:
            setattr(self, self.SAVE_CONTACT_ATTR, kwargs.pop(self.SAVE_CONTACT_ATTR))
        if self.SAVE_LABELS_ATTR in kwargs:
            setattr(self, self.SAVE_LABELS_ATTR, kwargs.pop(self.SAVE_LABELS_ATTR))

        super(Message, self).__init__(*args, **kwargs)

    @classmethod
    def get_unhandled(cls, org):
        return cls.objects.filter(org=org, is_handled=False)

    @classmethod
    def lock(cls, org, backend_id):
        return get_redis_connection().lock(MESSAGE_LOCK_KEY % (org.pk, backend_id), timeout=60)

    @classmethod
    def search(cls, org, user, search, modified_after=None, all=False):
        """
        Search for messages
        """
        folder = search.get("folder")
        label_id = search.get("label")
        text = search.get("text")
        contact_id = search.get("contact")
        after = search.get("after")
        before = search.get("before")

        is_admin = user.can_administer(org)
        partner = user.get_partner(org)
        all_label_access = is_admin or (partner and not partner.is_restricted)

        assert folder != MessageFolder.unlabelled or is_admin, "non-admin access to unlabelled messages"

        # track what we need to filter on by where is can be found in the database
        msg_filtering = {}
        lbl_filtering = {}
        ordering = ("-created_on",)

        # if this is a refresh we want everything with new actions and locks
        if modified_after:
            msg_filtering["modified_on__gt"] = modified_after

            # these are sorted by the UI again so we use the optimal ordering for the index here
            ordering = ("-modified_on", "-created_on")
        else:
            if folder == MessageFolder.inbox or folder == MessageFolder.unlabelled:
                lbl_filtering["is_archived"] = False
            elif folder == MessageFolder.flagged:
                lbl_filtering["is_flagged"] = True
                lbl_filtering["is_archived"] = False
            elif folder == MessageFolder.flagged_with_archived:
                lbl_filtering["is_flagged"] = True
            elif folder == MessageFolder.archived:
                lbl_filtering["is_archived"] = True

        if text:
            msg_filtering["text__icontains"] = text
            msg_filtering["created_on__gt"] = now() - timedelta(days=cls.SEARCH_BY_TEXT_DAYS)
        if contact_id:
            msg_filtering["contact__id"] = contact_id
        if after and not modified_after:
            lbl_filtering["created_on__gt"] = after
        if before:
            lbl_filtering["created_on__lt"] = before

        # only show non-deleted handled messages
        msgs = org.incoming_messages.filter(is_active=True, is_handled=True).order_by(*ordering)

        # handle views that don't require filtering by any labels
        if folder == MessageFolder.unlabelled:
            return (
                msgs.filter(type=Message.TYPE_INBOX, has_labels=False).filter(**msg_filtering).filter(**lbl_filtering)
            )
        if all_label_access and not label_id:
            if folder == MessageFolder.inbox:
                return msgs.filter(has_labels=True).filter(**msg_filtering).filter(**lbl_filtering)
            else:
                return msgs.filter(**msg_filtering).filter(**lbl_filtering)

        # we either want messages with a specific label or any of the labels this user has access to
        labels = Label.get_all(org, user)
        if label_id:
            labels = labels.filter(id=label_id)

        # if we're only filtering on things on the labelling table..
        if not msg_filtering and not all:
            lbl_filtering = {f"message_{k}": v for k, v in lbl_filtering.items()}
            message_ids = set()

            for label in labels:
                message_ids.update(
                    Labelling.objects.filter(label=label, **lbl_filtering)
                    .order_by("-message_created_on")
                    .values_list("message_id", flat=True)[:1000]
                )

            return Message.objects.filter(id__in=message_ids, is_handled=True).order_by("-created_on")

        msg_filtering["has_labels"] = True
        msgs = msgs.filter(labels__in=list(labels))

        return msgs.filter(**msg_filtering).filter(**lbl_filtering)

    def get_history(self):
        """
        Gets the actions for this message in reverse chronological order
        :return: the actions
        """
        return self.actions.select_related("created_by", "label").order_by("-pk")

    def get_lock(self, user):
        if self.locked_by_id and self.locked_on and self.locked_on > (now() - timedelta(seconds=MESSAGE_LOCK_SECONDS)):
            if self.locked_by_id != user.id:
                diff = (self.locked_on + timedelta(seconds=MESSAGE_LOCK_SECONDS)) - now()
                return diff.seconds

        return False

    def release(self):
        """
        Deletes this message, removing it from any labels (only callable by sync)
        """
        self.labels.clear()

        self.is_active = False
        self.modified_on = now()
        self.save(update_fields=("is_active", "modified_on"))

    def label(self, *labels):
        """
        Adds the given labels to this message
        """
        from casepro.profiles.models import Notification
        from casepro.statistics.models import DailyCount, datetime_to_date

        existing_label_ids = Labelling.objects.filter(message=self, label__in=labels).values_list("label", flat=True)
        add_labels = [l for l in labels if l.id not in existing_label_ids]
        new_labellings = [Labelling.create(l, self) for l in add_labels]
        Labelling.objects.bulk_create(new_labellings)

        day = datetime_to_date(self.created_on, self.org)
        for label in add_labels:
            DailyCount.record_item(day, DailyCount.TYPE_INCOMING, label)

        # notify all users who watch these labels
        for watcher in set(User.objects.filter(watched_labels__in=labels)):
            Notification.new_message_labelling(self.org, watcher, self)

    def unlabel(self, *labels):
        """
        Removes the given labels from this message
        """
        from casepro.statistics.models import DailyCount, datetime_to_date

        existing_labellings = Labelling.objects.filter(message=self, label__in=labels).select_related("label")

        day = datetime_to_date(self.created_on, self.org)
        for labelling in existing_labellings:
            DailyCount.record_removal(day, DailyCount.TYPE_INCOMING, labelling.label)

        Labelling.objects.filter(id__in=[l.id for l in existing_labellings]).delete()

    def clear_labels(self):
        """
        Removes all labels from this message
        """
        from casepro.statistics.models import DailyCount, datetime_to_date

        day = datetime_to_date(self.created_on, self.org)
        for label in self.labels.all():
            DailyCount.record_removal(day, DailyCount.TYPE_INCOMING, label)

        Labelling.objects.filter(message=self).delete()

    def update_labels(self, user, labels):
        """
        Updates this message's labels to match the given set, creating label and unlabel actions as necessary
        """
        with self.lock(self.org, self.backend_id):
            current_labels = set(self.labels.all())

            add_labels = [l for l in labels if l not in current_labels]
            rem_labels = [l for l in current_labels if l not in labels]

            for label in add_labels:
                self.bulk_label(self.org, user, [self], label)
            for label in rem_labels:
                self.bulk_unlabel(self.org, user, [self], label)

    def user_lock(self, user):
        """
        Marks this message as locked by a user
        """
        self.locked_on = now()
        self.locked_by = user
        self.modified_on = now()
        self.save(update_fields=("locked_on", "locked_by", "modified_on"))

    def user_unlock(self):
        """
        Marks this message as no longer locked by a user
        """
        self.locked_on = None
        self.locked_by = None
        self.modified_on = now()
        self.save(update_fields=("locked_on", "locked_by", "modified_on"))

    @staticmethod
    def bulk_flag(org, user, messages):
        messages = list(messages)
        if messages:
            org.incoming_messages.filter(org=org, pk__in=[m.pk for m in messages]).update(
                is_flagged=True, modified_on=now()
            )

            org.get_backend().flag_messages(org, messages)

            MessageAction.create(org, user, messages, MessageAction.FLAG)

    @staticmethod
    def bulk_unflag(org, user, messages):
        messages = list(messages)
        if messages:
            org.incoming_messages.filter(org=org, pk__in=[m.pk for m in messages]).update(
                is_flagged=False, modified_on=now()
            )

            org.get_backend().unflag_messages(org, messages)

            MessageAction.create(org, user, messages, MessageAction.UNFLAG)

    @staticmethod
    def bulk_label(org, user, messages, label):
        messages = list(messages)
        if messages:
            for msg in messages:
                msg.label(label)

            org.incoming_messages.filter(org=org, pk__in=[m.pk for m in messages]).update(modified_on=now())

            if label.is_synced:
                org.get_backend().label_messages(org, messages, label)

            MessageAction.create(org, user, messages, MessageAction.LABEL, label)

    @staticmethod
    def bulk_unlabel(org, user, messages, label):
        messages = list(messages)
        if messages:
            for msg in messages:
                msg.unlabel(label)

            org.incoming_messages.filter(org=org, pk__in=[m.pk for m in messages]).update(modified_on=now())

            if label.is_synced:
                org.get_backend().unlabel_messages(org, messages, label)

            MessageAction.create(org, user, messages, MessageAction.UNLABEL, label)

    @staticmethod
    def bulk_archive(org, user, messages):
        messages = list(messages)
        if messages:
            org.incoming_messages.filter(org=org, pk__in=[m.pk for m in messages]).update(
                is_archived=True, modified_on=now()
            )

            org.get_backend().archive_messages(org, messages)

            MessageAction.create(org, user, messages, MessageAction.ARCHIVE)

    @staticmethod
    def bulk_restore(org, user, messages):
        messages = list(messages)
        if messages:
            org.incoming_messages.filter(org=org, pk__in=[m.pk for m in messages]).update(
                is_archived=False, modified_on=now()
            )

            org.get_backend().restore_messages(org, messages)

            MessageAction.create(org, user, messages, MessageAction.RESTORE)

    def as_json(self):
        """
        Prepares this message for JSON serialization
        """
        return {
            "id": self.backend_id,
            "contact": self.contact.as_json(full=False),
            "text": self.text,
            "time": self.created_on,
            "labels": [l.as_json(full=False) for l in self.labels.all()],
            "flagged": self.is_flagged,
            "archived": self.is_archived,
            "flow": self.type == self.TYPE_FLOW,
            "case": self.case.as_json(full=False) if self.case else None,
        }

    def __str__(self):
        return self.text if self.text else self.pk class BaseCasesTest(DashTest):
    """
    Base class for all test cases
    """

    def setUp(self):
        super(BaseCasesTest, self).setUp()

        settings.SITE_BACKEND = "casepro.test.TestBackend"
        settings.SITE_ORGS_STORAGE_ROOT = "test_orgs"

        # some orgs
        self.unicef = self.create_org("UNICEF", timezone=zoneinfo.ZoneInfo("Africa/Kampala"), subdomain="unicef")
        self.nyaruka = self.create_org("Nyaruka", timezone=zoneinfo.ZoneInfo("Africa/Kigali"), subdomain="nyaruka")

        # some admins for those orgs
        self.admin = self.create_admin(self.unicef, "Kidus", "kidus@unicef.org")
        self.norbert = self.create_admin(self.nyaruka, "Norbert Kwizera", "norbert@nyaruka.com")

        # some message labels
        self.aids = self.create_label(self.unicef, "L-001", "AIDS", "Messages about AIDS", ["aids", "hiv"])
        self.pregnancy = self.create_label(
            self.unicef, "L-002", "Pregnancy", "Messages about pregnancy", ["pregnant", "pregnancy"]
        )
        self.tea = self.create_label(self.unicef, None, "Tea", "Messages about tea", ["tea", "chai"], is_synced=False)
        self.code = self.create_label(self.nyaruka, "L-101", "Code", "Messages about code", ["java", "python", "go"])

        # some message faqs
        self.preg_faq1_eng = self.create_faq(
            self.unicef, "How do I know I'm pregnant?", "Do a pregnancy test.", "eng", None, [self.pregnancy]
        )
        self.preg_faq1_zul = self.create_faq(
            self.unicef, "ZUL How do I know I'm pregnant?", "ZUL Do a pregnancy test.", "zul", self.preg_faq1_eng, []
        )
        self.preg_faq1_lug = self.create_faq(
            self.unicef, "LUG How do I know I'm pregnant?", "LUG Do a pregnancy test.", "lug", self.preg_faq1_eng, []
        )
        self.preg_faq2_eng = self.create_faq(
            self.unicef,
            "How do I prevent HIV transfer to my baby?",
            "Take ARVs.",
            "eng",
            None,
            [self.pregnancy, self.aids],
        )
        self.tea_faq1_eng = self.create_faq(
            self.unicef, "Does tea contain caffeine?", "It varies - black tea does.", "eng", None, [self.tea]
        )

        # some partners
        self.moh = self.create_partner(self.unicef, "MOH", "Ministry of Health", None, [self.aids, self.pregnancy])
        self.who = self.create_partner(self.unicef, "WHO", "World Health Organisation", None, [self.aids])
        self.klab = self.create_partner(self.nyaruka, "kLab", "Kigali Lab", None, [self.code])

        # some users in those partners
        self.user1 = self.create_user(self.unicef, self.moh, ROLE_MANAGER, "Evan", "evan@unicef.org")
        self.user2 = self.create_user(self.unicef, self.moh, ROLE_ANALYST, "Rick", "rick@unicef.org")
        self.user3 = self.create_user(self.unicef, self.who, ROLE_MANAGER, "Carol", "carol@unicef.org")
        self.user4 = self.create_user(self.nyaruka, self.klab, ROLE_ANALYST, "Bosco", "bosco@klab.rw")

        # some groups
        self.males = self.create_group(self.unicef, "G-001", "Males")
        self.females = self.create_group(self.unicef, "G-002", "Females")
        self.reporters = self.create_group(self.unicef, "G-003", "Reporters", suspend_from=True, is_visible=False)
        self.registered = self.create_group(self.unicef, "G-004", "Registered (Dynamic)", is_dynamic=True)
        self.coders = self.create_group(self.nyaruka, "G-005", "Coders")

        # some fields
        self.nickname = self.create_field(self.unicef, "nickname", "Nickname", value_type="T")
        self.age = self.create_field(self.unicef, "age", "Age", value_type="N")
        self.state = self.create_field(self.unicef, "state", "State", value_type="S", is_visible=False)
        self.motorbike = self.create_field(self.nyaruka, "motorbike", "Moto", value_type="T")

    def create_org(self, name, timezone, subdomain):
        return Org.objects.create(
            name=name, timezone=timezone, subdomain=subdomain, created_by=self.superuser, modified_by=self.superuser
        )

    def create_partner(self, org, name, description, primary_contact, labels=(), restricted=True):
        return Partner.create(org, name, description, primary_contact, restricted, labels, None)

    def create_admin(self, org, name, email):
        return Profile.create_org_user(org, name, email, email)

    def create_user(self, org, partner, role, name, email):
        return Profile.create_partner_user(org, partner, role, name, email, email)

    def create_label(self, org, uuid, name, description, keywords=(), **kwargs):
        label = Label.objects.create(org=org, uuid=uuid, name=name, description=description, **kwargs)

        if keywords:
            rule = Rule.create(org, [ContainsTest(keywords, Quantifier.ANY)], [LabelAction(label)])
            label.rule = rule
            label.save(update_fields=("rule",))

        return label

    def create_rule(self, org, tests, actions):
        return Rule.create(org, tests, actions)

    def create_faq(self, org, question, answer, language, parent, labels=(), **kwargs):
        faq = FAQ.create(org, question, answer, language, parent, labels)
        return faq

    def create_contact(self, org, uuid, name, groups=(), fields=None, is_stub=False):
        contact = Contact.objects.create(org=org, uuid=uuid, name=name, is_stub=is_stub, fields=fields, language="eng")
        contact.groups.add(*groups)
        return contact

    def create_group(self, org, uuid, name, count=None, is_dynamic=False, is_visible=True, suspend_from=False):
        return Group.objects.create(
            org=org,
            uuid=uuid,
            name=name,
            count=count,
            is_dynamic=is_dynamic,
            is_visible=is_visible,
            suspend_from=suspend_from,
        )

    def create_field(self, org, key, label, value_type="T", is_visible=True):
        return Field.objects.create(org=org, key=key, label=label, value_type=value_type, is_visible=is_visible)

    def create_message(self, org, backend_id, contact, text, labels=(), **kwargs):
        if "type" not in kwargs:
            kwargs["type"] = "I"
        if "created_on" not in kwargs:
            kwargs["created_on"] = now()

        msg = Message.objects.create(org=org, backend_id=backend_id, contact=contact, text=text, **kwargs)
        msg.label(*labels)
        return msg

    def create_outgoing(self, org, user, broadcast_id, activity, text, contact, **kwargs):
        return Outgoing.objects.create(
            org=org,
            partner=user.get_partner(org),
            backend_broadcast_id=broadcast_id,
            activity=activity,
            text=text,
            contact=contact,
            created_by=user,
            **kwargs,
        )

    def create_case(self, org, contact, assignee, message, labels=(), user_assignee=None, **kwargs):
        case = Case.objects.create(
            org=org,
            contact=contact,
            assignee=assignee,
            user_assignee=user_assignee,
            initial_message=message,
            **kwargs,
        )
        case.labels.add(*labels)

        if "opened_on" in kwargs:  # uses auto_now_add
            case.opened_on = kwargs["opened_on"]
            case.save(update_fields=("opened_on",))

        if message:
            message.case = case
            message.save(update_fields=("case",))

        return case

    def url_post_json(self, subdomain, url, data):
        return self.url_post(subdomain, url, json.dumps(data), content_type="application/json")

    def assertLoginRedirect(self, response, next_url):
        self.assertRedirects(response, "/users/login/?next=%s" % next_url)

    def assertNotCalled(self, mock):
        """
        Because mock.assert_not_called doesn't exist in Python 2
        """
        self.assertEqual(len(mock.mock_calls), 0, "Expected no calls, called %d times" % len(mock.mock_calls))

    def openWorkbook(self, filename):
        return open_workbook("%s/%s" % (settings.MEDIA_ROOT, filename), "rb")

    def assertExcelRow(self, sheet, row_num, values, tz=None):
        """
        Asserts the cell values in the given worksheet row. Date values are converted using the provided timezone.
        """
        expected_values = []
        for expected in values:
            # if expected value is datetime, localize and remove microseconds
            if isinstance(expected, datetime):
                expected = expected.astimezone(tz).replace(microsecond=0, tzinfo=None)
            elif isinstance(expected, date):
                expected = datetime.combine(expected, time(0, 0))

            expected_values.append(expected)

        actual_values = []
        for c in range(0, sheet.ncols):
            cell = sheet.cell(row_num, c)
            actual = cell.value

            if cell.ctype == XL_CELL_DATE:
                actual = datetime(*xldate_as_tuple(actual, sheet.book.datemode))

            actual_values.append(actual)

        self.assertEqual(actual_values, expected_values)

    def assertSentMail(self, recipients, reset_outbox=True):
        self.assertEqual([e.to[0] for e in mail.outbox], recipients)

        if reset_outbox:
            mail.outbox = []

    def login(self, user):
        password = "root" if user == self.superuser else user.username
        result = self.client.login(username=user.username, password=password)
        self.assertTrue(result, "Couldn't login as %(user)s / %(user)s" % dict(user=user.username)) class Action(object):
    """
    An action which can be performed on a message
    """

    __metaclass__ = ABCMeta

    TYPE = None
    CLASS_BY_TYPE = None  # lazily initialized below

    @classmethod
    def from_json(cls, json_obj, context):
        if not cls.CLASS_BY_TYPE:
            cls.CLASS_BY_TYPE = {
                LabelAction.TYPE: LabelAction,
                FlagAction.TYPE: FlagAction,
                ArchiveAction.TYPE: ArchiveAction,
            }

        action_type = json_obj["type"]
        action_cls = cls.CLASS_BY_TYPE.get(action_type)
        if not action_cls:  # pragma: no cover
            raise ValueError("Unknown action type: %s" % action_type)

        return action_cls.from_json(json_obj, context)

    @abstractmethod
    def to_json(self):  # pragma: no cover
        pass

    @abstractmethod
    def get_description(self):  # pragma: no cover
        pass

    def __eq__(self, other):
        return self.TYPE == other.TYPE

    def __ne__(self, other):
        return not self.__eq__(other)

    def __hash__(self):
        return hash(self.TYPE) class ArchiveAction(Action):
    """
    Archives the message
    """

    TYPE = "archive"

    @classmethod
    def from_json(cls, json_obj, context):
        return cls()

    def to_json(self):
        return {"type": self.TYPE}

    def get_description(self):
        return "archive"

    def apply_to(self, org, messages):
        Message.objects.filter(pk__in=[m.pk for m in messages]).update(is_archived=True)

        try:
            org.get_backend().archive_messages(org, messages)
        except Exception as e:
            logger.exception(e) class ContainsTest(Test):
    """
    Test that returns whether the message text contains or doesn't contain the given keywords
    """

    TYPE = "contains"

    def __init__(self, keywords, quantifier):
        self.keywords = [normalize(word) for word in keywords]
        self.quantifier = quantifier

    @classmethod
    def from_json(cls, json_obj, context):
        return cls(json_obj["keywords"], Quantifier.from_json(json_obj["quantifier"]))

    def to_json(self):
        return {"type": self.TYPE, "keywords": self.keywords, "quantifier": self.quantifier.to_json()}

    def get_description(self):
        quoted_keywords = ['"%s"' % w for w in self.keywords]
        return "message contains %s %s" % (str(self.quantifier), ", ".join(quoted_keywords))

    def matches(self, message):
        text = normalize(message.text)

        def keyword_check(w):
            return lambda: bool(regex.search(r"\b" + w + r"\b", text, flags=regex.UNICODE | regex.V0))

        checks = [keyword_check(keyword) for keyword in self.keywords]

        return self.quantifier.evaluate(checks)

    @classmethod
    def is_valid_keyword(cls, keyword):
        return KEYWORD_REGEX.match(keyword)

    def __eq__(self, other):
        return (
            other
            and self.TYPE == other.TYPE
            and self.keywords == other.keywords
            and self.quantifier == other.quantifier
        ) class DeserializationContext(object):
    """
    Context object passed to all test or action from_json methods
    """

    def __init__(self, org):
        self.org = org class FieldTest(Test):
    """
    Test that returns whether the message was sent from a contact with the given field value
    """

    TYPE = "field"

    def __init__(self, key, values):
        self.key = key
        self.values = [normalize(v) for v in values]

    @classmethod
    def from_json(cls, json_obj, context):
        return cls(json_obj["key"], json_obj["values"])

    def to_json(self):
        return {"type": self.TYPE, "key": self.key, "values": self.values}

    def get_description(self):
        quoted_values = ['"%s"' % v for v in self.values]
        return "contact.%s is %s %s" % (self.key, Quantifier.ANY, ", ".join(quoted_values))

    def matches(self, message):
        if message.contact.fields:
            contact_value = normalize(message.contact.fields.get(self.key, ""))

            for value in self.values:
                if value == contact_value:
                    return True
        return False

    def __eq__(self, other):
        return other and self.TYPE == other.TYPE and self.key == other.key and self.values == other.values class FlagAction(Action):
    """
    Flags the message
    """

    TYPE = "flag"

    @classmethod
    def from_json(cls, json_obj, context):
        return cls()

    def to_json(self):
        return {"type": self.TYPE}

    def get_description(self):
        return "flag"

    def apply_to(self, org, messages):
        Message.objects.filter(pk__in=[m.pk for m in messages]).update(is_flagged=True)

        try:
            org.get_backend().flag_messages(org, messages)
        except Exception as e:
            logger.exception(e) class GroupsTest(Test):
    """
    Test that returns whether the message was sent from the given contact groups
    """

    TYPE = "groups"

    def __init__(self, groups, quantifier):
        self.groups = groups
        self.quantifier = quantifier

    @classmethod
    def from_json(cls, json_obj, context):
        groups = list(Group.objects.filter(org=context.org, pk__in=json_obj["groups"]).order_by("pk"))
        return cls(groups, Quantifier.from_json(json_obj["quantifier"]))

    def to_json(self):
        return {"type": self.TYPE, "groups": [g.pk for g in self.groups], "quantifier": self.quantifier.to_json()}

    def get_description(self):
        group_names = [g.name for g in self.groups]
        return "contact belongs to %s %s" % (str(self.quantifier), ", ".join(group_names))

    def matches(self, message):
        contact_groups = set(message.contact.groups.all())

        def group_check(g):
            return lambda: g in contact_groups

        checks = [group_check(group) for group in self.groups]

        return self.quantifier.evaluate(checks)

    def __eq__(self, other):
        return (
            other and self.TYPE == other.TYPE and self.groups == other.groups and self.quantifier == other.quantifier
        ) class LabelAction(Action):
    """
    Adds a label to the message
    """

    TYPE = "label"

    def __init__(self, label):
        self.label = label

    @classmethod
    def from_json(cls, json_obj, context):
        return cls(Label.objects.get(org=context.org, pk=json_obj["label"]))

    def to_json(self):
        return {"type": self.TYPE, "label": self.label.pk}

    def get_description(self):
        return "apply label '%s'" % self.label.name

    def apply_to(self, org, messages):
        for msg in messages:
            msg.label(self.label)

        if self.label.is_synced:
            try:
                org.get_backend().label_messages(org, messages, self.label)
            except Exception as e:
                logger.exception(e)

    def __eq__(self, other):
        return self.TYPE == other.TYPE and self.label == other.label

    def __hash__(self):
        return hash(self.TYPE + str(self.label.pk)) class Quantifier(Enum):
    """
    Tests are typically composed of multiple conditions, e.g. contains ANY of X, Y or Z.
    """

    NONE = (1, _("none of"))
    ANY = (2, _("any of"))
    ALL = (3, _("all of"))

    def __init__(self, val, text):
        self.val = val
        self.text = text

    @classmethod
    def from_json(cls, val):
        return cls[val.upper()]

    def to_json(self):
        return self.name.lower()

    def evaluate(self, condition_callables):
        if self == Quantifier.NONE:
            for condition in condition_callables:
                if condition():
                    return False
            return True
        elif self == Quantifier.ANY:
            for condition in condition_callables:
                if condition():
                    return True
            return False
        elif self == Quantifier.ALL:
            for condition in condition_callables:
                if not condition():
                    return False
            return True

    def __str__(self):
        return str(self.text) class Rule(models.Model):
    """
    At some point this will become a first class object, but for now it is always attached to a label.
    """

    org = models.ForeignKey(Org, verbose_name=_("Organization"), related_name="rules", on_delete=models.PROTECT)

    tests = models.TextField()

    actions = models.TextField()

    @classmethod
    def create(cls, org, tests, actions):
        return cls.objects.create(org=org, tests=json_encode(tests), actions=json_encode(actions))

    @classmethod
    def get_all(cls, org):
        return org.rules.all()

    def get_tests(self):
        return get_obj_cacheable(self, "_tests", lambda: self._get_tests())

    def _get_tests(self):
        return [Test.from_json(t, DeserializationContext(self.org)) for t in json.loads(self.tests)]

    def get_tests_description(self):
        return _(" and ").join([t.get_description() for t in self.get_tests()])

    def get_actions(self):
        return get_obj_cacheable(self, "_actions", lambda: self._get_actions())

    def _get_actions(self):
        return [Action.from_json(a, DeserializationContext(self.org)) for a in json.loads(self.actions)]

    def get_actions_description(self):
        return _(" and ").join([a.get_description() for a in self.get_actions()])

    def matches(self, message):
        """
        Returns whether this rule matches the given message, i.e. all of its tests match the message
        """
        for test in self.get_tests():
            if not test.matches(message):
                return False
        return True

    class BatchProcessor(object):
        """
        Applies a set of rules to a batch of messages in a way that allows same actions to be merged and reduces needed
        calls to the backend.
        """

        def __init__(self, org, rules):
            self.org = org
            self.rules = rules
            self.messages_by_action = defaultdict(set)

        def include_messages(self, *messages):
            """
            Includes the given messages in this batch processing
            :param messages: the messages to include
            :return: tuple of the number of rules matched, and the number of actions that will be performed
            """
            num_rules_matched = 0
            num_actions_deferred = 0

            for message in messages:
                for rule in self.rules:
                    if rule.matches(message):
                        num_rules_matched += 1
                        for action in rule.get_actions():
                            self.messages_by_action[action].add(message)
                            num_actions_deferred += 1

            return num_rules_matched, num_actions_deferred

        def apply_actions(self):
            """
            Applies the actions gathered by this processor
            """
            for action, messages in self.messages_by_action.items():
                action.apply_to(self.org, messages) class Test(object):
    """
    A test which can be evaluated to true or false on a given message
    """

    __metaclass__ = ABCMeta

    TYPE = None
    CLASS_BY_TYPE = None  # lazily initialized below

    @classmethod
    def from_json(cls, json_obj, context):
        if not cls.CLASS_BY_TYPE:
            cls.CLASS_BY_TYPE = {
                ContainsTest.TYPE: ContainsTest,
                WordCountTest.TYPE: WordCountTest,
                GroupsTest.TYPE: GroupsTest,
                FieldTest.TYPE: FieldTest,
            }

        test_type = json_obj["type"]
        test_cls = cls.CLASS_BY_TYPE.get(test_type, None)
        if not test_cls:  # pragma: no cover
            raise ValueError("Unknown test type: %s" % test_type)

        return test_cls.from_json(json_obj, context)

    @abstractmethod
    def to_json(self):  # pragma: no cover
        pass

    @abstractmethod
    def get_description(self):  # pragma: no cover
        pass

    @abstractmethod
    def matches(self, message):
        """
        Subclasses must implement this to return a boolean.
        """

    def __eq__(self, other):  # pragma: no cover
        return other and self.TYPE == other.TYPE

    def __ne__(self, other):
        return not self.__eq__(other) class WordCountTest(Test):
    """
    Test that returns whether the message text contains at least the given number of words
    """

    TYPE = "words"

    def __init__(self, minimum):
        self.minimum = minimum

    @classmethod
    def from_json(cls, json_obj, context):
        return cls(json_obj["minimum"])

    def to_json(self):
        return {"type": self.TYPE, "minimum": self.minimum}

    def get_description(self):
        return "message has at least %d words" % self.minimum

    def matches(self, message):
        num_words = len(regex.findall(r"\w+", message.text, flags=regex.UNICODE | regex.V0))
        return num_words >= self.minimum

    def __eq__(self, other):
        return other and self.TYPE == other.TYPE and self.minimum == other.minimum @register.filter
def render_actions(rule):
    rendered_actions = []
    for action in rule.get_actions():
        renderer = ACTION_RENDERERS[type(action)]
        rendered_actions.append(renderer(action))

    return mark_safe(_(", and ").join(rendered_actions)) @register.filter
def render_tests(rule):
    rendered_tests = []
    for test in rule.get_tests():
        renderer = TEST_RENDERERS[type(test)]
        rendered_tests.append(renderer(test))

    return mark_safe(_(", and ").join(rendered_tests))
from unittest.mock import call, patch
from django.urls import reverse
from casepro.msgs.models import Message
from casepro.test import BaseCasesTest
from .models import (
    Action,
    ArchiveAction,
    ContainsTest,
    DeserializationContext,
    FieldTest,
    FlagAction,
    GroupsTest,
    LabelAction,
    Quantifier,
    Rule,
    Test,
    WordCountTest,
)
from .templatetags.rules import render_actions, render_tests
        self.assertEqual(test.matches(msg), result)

    def test_contains(self):
        test = Test.from_json({"type": "contains", "keywords": ["RED", "Blue"], "quantifier": "any"}, self.context)
        self.assertEqual(test.TYPE, "contains")
        self.assertEqual(test.keywords, ["red", "blue"])
        self.assertEqual(test.quantifier, Quantifier.ANY)
        self.assertEqual(test.to_json(), {"type": "contains", "keywords": ["red", "blue"], "quantifier": "any"})
        self.assertEqual(test.get_description(), 'message contains any of "red", "blue"')

        self.assertEqual(test, ContainsTest(["RED", "Blue"], Quantifier.ANY))
        self.assertNotEqual(test, ContainsTest(["RED", "Green"], Quantifier.ANY))

        self.assertTest(test, self.ann, "Fred Blueth", False)
        self.assertTest(test, self.ann, "red", True)
        self.assertTest(test, self.ann, "tis Blue", True)

        test.quantifier = Quantifier.ALL

        self.assertTest(test, self.ann, "Fred Blueth", False)
        self.assertTest(test, self.ann, "red", False)
        self.assertTest(test, self.ann, "yo RED Blue", True)

        test.quantifier = Quantifier.NONE

        self.assertTest(test, self.ann, "Fred Blueth", True)
        self.assertTest(test, self.ann, "red", False)
        self.assertTest(test, self.ann, "yo RED Blue", False)

    def test_word_count(self):
        test = Test.from_json({"type": "words", "minimum": 2}, self.context)
        self.assertEqual(test.TYPE, "words")
        self.assertEqual(test.minimum, 2)
        self.assertEqual(test.to_json(), {"type": "words", "minimum": 2})
        self.assertEqual(test.get_description(), "message has at least 2 words")

        self.assertEqual(test, WordCountTest(2))
        self.assertNotEqual(test, WordCountTest(3))

        self.assertTest(test, self.ann, "!", False)
        self.assertTest(test, self.ann, "no!", False)
        self.assertTest(test, self.ann, "ok  maybe ", True)
        self.assertTest(test, self.ann, "uh-ok-sure", True)

    def test_groups(self):
        test = Test.from_json(
            {"type": "groups", "groups": [self.females.pk, self.reporters.pk], "quantifier": "any"}, self.context
        )
        self.assertEqual(test.TYPE, "groups")
        self.assertEqual(set(test.groups), {self.females, self.reporters})
        self.assertEqual(test.quantifier, Quantifier.ANY)
        self.assertEqual(
            test.to_json(), {"type": "groups", "groups": [self.females.pk, self.reporters.pk], "quantifier": "any"}
        )
        self.assertEqual(test.get_description(), "contact belongs to any of Females, Reporters")

        self.assertTest(test, self.ann, "Yes", True)
        self.assertTest(test, self.bob, "Yes", False)
        self.assertTest(test, self.cat, "Yes", True)

        test.quantifier = Quantifier.ALL

        self.assertTest(test, self.ann, "Yes", False)
        self.assertTest(test, self.bob, "Yes", False)
        self.assertTest(test, self.cat, "Yes", True)

        test.quantifier = Quantifier.NONE

        self.assertTest(test, self.ann, "Yes", False)
        self.assertTest(test, self.bob, "Yes", True)
        self.assertTest(test, self.cat, "Yes", False)

    def test_field(self):
        test = Test.from_json({"type": "field", "key": "city", "values": ["Kigali", "Lusaka"]}, self.context)
        self.assertEqual(test.TYPE, "field")
        self.assertEqual(test.key, "city")
        self.assertEqual(test.values, ["kigali", "lusaka"])
        self.assertEqual(test.to_json(), {"type": "field", "key": "city", "values": ["kigali", "lusaka"]})
        self.assertEqual(test.get_description(), 'contact.city is any of "kigali", "lusaka"')

        self.assertTest(test, self.ann, "Yes", True)
        self.assertTest(test, self.bob, "Yes", False)
        self.assertTest(test, self.cat, "Yes", False)

    def test_is_valid_keyword(self):
        self.assertTrue(ContainsTest.is_valid_keyword("tú"))
        self.assertTrue(ContainsTest.is_valid_keyword("kit"))
        self.assertTrue(ContainsTest.is_valid_keyword("kit-kat"))
        self.assertTrue(ContainsTest.is_valid_keyword("kit kat"))
        self.assertTrue(ContainsTest.is_valid_keyword("kit-kat wrapper"))

        self.assertFalse(ContainsTest.is_valid_keyword("i"))  # too short
        self.assertFalse(ContainsTest.is_valid_keyword(" kitkat"))  # can't start with a space
        self.assertFalse(ContainsTest.is_valid_keyword("-kit"))  # can't start with a dash
        self.assertFalse(ContainsTest.is_valid_keyword("kat "))  # can't end with a space
        self.assertFalse(ContainsTest.is_valid_keyword("kat-"))  # can't end with a dash


class RulesTemplateTagsTest(BaseCasesTest):
    def setUp(self):
        super(RulesTemplateTagsTest, self).setUp()

        self.context = DeserializationContext(self.unicef)

    def test_render_contains_test(self):
        # single keyword
        test = Test.from_json({"type": "contains", "keywords": ["blue"], "quantifier": "all"}, self.context)
        rule = Rule.create(self.unicef, [test], [])
        self.assertEqual('message contains <i>"blue"</i>', render_tests(rule))

        # multiple keywords
        test = Test.from_json({"type": "contains", "keywords": ["RED", "Blue"], "quantifier": "any"}, self.context)
        rule = Rule.create(self.unicef, [test], [])
        self.assertEqual('message contains any of <i>"red"</i>, <i>"blue"</i>', render_tests(rule))

        # check that keywords are correctly escaped to prevent XSS attacks
        test.keywords = ["<script>alert('Hi')</script>"]
        rule = Rule.create(self.unicef, [test], [])

        self.assertEqual(

            'message contains <i>"&lt;script&gt;alert(&#x27;hi&#x27;)&lt;/script&gt;"</i>', render_tests(rule)