changedetection.io/changedetectionio/tests/unit/test_conditions.py

from changedetectionio.conditions import execute_ruleset_against_all_plugins
from changedetectionio.model import CONDITIONS_MATCH_LOGIC_DEFAULT
from changedetectionio.store import ChangeDetectionStore
import shutil
import tempfile
import time
import unittest
import uuid


class TestTriggerConditions(unittest.TestCase):
    def setUp(self):

        # Create a temporary directory for the test datastore
        self.test_datastore_path = tempfile.mkdtemp()

        # Initialize ChangeDetectionStore with our test path and no default watches
        self.store = ChangeDetectionStore(
            datastore_path=self.test_datastore_path,
            include_default_watches=False
        )

        # Add a test watch
        watch_url = "https://example.com"
        self.watch_uuid = self.store.add_watch(url=watch_url)

    def tearDown(self):
      # Clean up the test datastore
      self.store.stop_thread = True
      time.sleep(0.5)  # Give thread time to stop
      shutil.rmtree(self.test_datastore_path)

    def test_conditions_execution_pass(self):
        # Get the watch object
        watch = self.store.data['watching'][self.watch_uuid]

        # Create and save a snapshot
        first_content = "I saw 100 people at a rock show"
        timestamp1 = int(time.time())
        snapshot_id1 = str(uuid.uuid4())
        watch.save_history_blob(contents=first_content,
                                timestamp=timestamp1,
                                snapshot_id=snapshot_id1)

        # Add another snapshot
        second_content = "I saw 200 people at a rock show"
        timestamp2 = int(time.time()) + 60
        snapshot_id2 = str(uuid.uuid4())
        watch.save_history_blob(contents=second_content,
                                timestamp=timestamp2,
                                snapshot_id=snapshot_id2)

        # Verify both snapshots are stored
        history = watch.history
        self.assertEqual(len(history), 2)

        # Retrieve and check snapshots
        #snapshot1 = watch.get_history_snapshot(timestamp=str(timestamp1))
        #snapshot2 = watch.get_history_snapshot(timestamp=str(timestamp2))

        self.store.data['watching'][self.watch_uuid].update(
            {
                "conditions_match_logic": CONDITIONS_MATCH_LOGIC_DEFAULT,
                "conditions": [
                    {"operator": ">=", "field": "extracted_number", "value": "10"},
                    {"operator": "<=", "field": "extracted_number", "value": "5000"},
                    {"operator": "in", "field": "page_filtered_text", "value": "rock"},
                    #{"operator": "starts_with", "field": "page_text", "value": "I saw"},
                ]
            }
        )

        # ephemeral_data - some data that could exist before the watch saved a new version
        result = execute_ruleset_against_all_plugins(current_watch_uuid=self.watch_uuid,
                                                     application_datastruct=self.store.data,
                                                     ephemeral_data={'text': "I saw 500 people at a rock show"})

        # @todo - now we can test that 'Extract number' increased more than X since last time
        self.assertTrue(result.get('result'))



    def test_conditions_filter_complete_rules_with_zero_values(self):
        from changedetectionio.conditions import filter_complete_rules

        rules = [
            {"operator": "==", "field": "word_count", "value": 0},
            {"operator": "<=", "field": "levenshtein_distance", "value": 0.0},
            {"operator": "==", "field": "price", "value": "0"},
            {"operator": "==", "field": "empty_val", "value": ""},
            {"operator": "==", "field": "none_val", "value": None},
            {"operator": "==", "field": "str_none", "value": "None"},
            {"operator": "", "field": "missing_op", "value": 0},
            {"operator": "None", "field": "str_none_op", "value": 0},
            {"operator": "==", "field": "", "value": 0},
        ]
        complete = filter_complete_rules(rules)
        self.assertEqual(len(complete), 3)
        self.assertEqual(complete[0]["value"], 0)
        self.assertEqual(complete[1]["value"], 0.0)
        self.assertEqual(complete[2]["value"], "0")

    def test_conditions_convert_to_jsonlogic_with_zero_values(self):
        from json_logic import jsonLogic
        from changedetectionio.conditions import convert_to_jsonlogic

        rule_int_zero = [{"operator": "==", "field": "word_count", "value": 0}]
        jl_int = convert_to_jsonlogic("and", rule_int_zero)
        self.assertEqual(jl_int, {"==": [{"var": "word_count"}, 0]})
        self.assertTrue(jsonLogic(jl_int, {"word_count": 0}))
        self.assertFalse(jsonLogic(jl_int, {"word_count": 5}))

        rule_float_zero = [{"operator": "<=", "field": "levenshtein_distance", "value": 0.0}]
        jl_float = convert_to_jsonlogic("and", rule_float_zero)
        self.assertEqual(jl_float, {"<=": [{"var": "levenshtein_distance"}, 0.0]})
        self.assertTrue(jsonLogic(jl_float, {"levenshtein_distance": 0.0}))
        self.assertFalse(jsonLogic(jl_float, {"levenshtein_distance": 2.5}))

    def test_conditions_execution_zero_word_count(self):
        # Test condition checking for empty page (word_count == 0)
        self.store.data['watching'][self.watch_uuid].update(
            {
                "conditions_match_logic": "ALL",
                "conditions": [
                    {"operator": "==", "field": "word_count", "value": 0},
                ],
            }
        )
        # Empty text has word_count 0 -> condition should match (True)
        res_empty = execute_ruleset_against_all_plugins(
            current_watch_uuid=self.watch_uuid,
            application_datastruct=self.store.data,
            ephemeral_data={'text': ""},
        )
        self.assertTrue(res_empty.get('result'))

        # Non-empty text has word_count > 0 -> condition should not match (False)
        res_nonempty = execute_ruleset_against_all_plugins(
            current_watch_uuid=self.watch_uuid,
            application_datastruct=self.store.data,
            ephemeral_data={'text': "Hello world"},
        )
        self.assertFalse(res_nonempty.get('result'))


if __name__ == '__main__':
    unittest.main()