Merge pull request #540 from robbrad/headless_refactor

Headless refactor

custom_components/uk_bin_collection/config_flow.py

@@ -29,8 +29,10 @@ async def get_council_schema(self, council=str) -> vol.Schema:
         if self.councils_data is None:
             self.councils_data = await self.get_councils_json()
         council_schema = vol.Schema({})
-        if ("skip_get_url" not in self.councils_data[council] or
-                "custom_component_show_url_field" in self.councils_data[council]):
+        if (
+            "skip_get_url" not in self.councils_data[council]
+            or "custom_component_show_url_field" in self.councils_data[council]
+        ):
             council_schema = council_schema.extend(
                 {vol.Required("url", default=""): cv.string}
             )
@@ -54,6 +56,9 @@ async def get_council_schema(self, council=str) -> vol.Schema:
             council_schema = council_schema.extend(
                 {vol.Required("web_driver", default=""): cv.string}
             )
+            council_schema = council_schema.extend(
+                {vol.Optional("headless", default=True): cv.boolean}
+            )
         return council_schema
 
     async def async_step_user(self, user_input=None):
@@ -116,9 +121,7 @@ async def async_step_council(self, user_input=None):
 
             # Create the config entry
             _LOGGER.info(LOG_PREFIX + "Creating config entry with data: %s", user_input)
-            return self.async_create_entry(
-                title=user_input["name"], data=user_input
-            )
+            return self.async_create_entry(title=user_input["name"], data=user_input)
 
         # Show the configuration form to the user with the specific councils necessary fields
         council_schema = await self.get_council_schema(self.data["council"])

custom_components/uk_bin_collection/const.py

@@ -15,4 +15,4 @@
 STATE_ATTR_NEXT_COLLECTION = "next_collection"
 STATE_ATTR_DAYS = "days"
 
-DEVICE_CLASS = "bin_collection_schedule"
+DEVICE_CLASS = "bin_collection_schedule"

custom_components/uk_bin_collection/sensor.py

@@ -73,6 +73,7 @@ async def async_setup_entry(
         for bin_type in coordinator.data.keys()
     )
 
+
 def get_latest_collection_info(data) -> dict:
     # Get the current date
     current_date = datetime.now()
@@ -92,14 +93,17 @@ def get_latest_collection_info(data) -> dict:
         if collection_date.date() >= current_date.date():
             # If the bin type is in the dict, update its collection date if needed; otherwise, add it
             if bin_type in next_collection_dates:
-                if collection_date < datetime.strptime(next_collection_dates[bin_type], "%d/%m/%Y"):
+                if collection_date < datetime.strptime(
+                    next_collection_dates[bin_type], "%d/%m/%Y"
+                ):
                     next_collection_dates[bin_type] = collection_date_str
             else:
                 next_collection_dates[bin_type] = collection_date_str
-                
+
     _LOGGER.info(f"{LOG_PREFIX} Next Collection Dates: {next_collection_dates}")
     return next_collection_dates
 
+
 class HouseholdBinCoordinator(DataUpdateCoordinator):
     """Household Bin Coordinator"""
 
@@ -124,8 +128,10 @@ async def _async_update_data(self):
             _LOGGER.info(f"{LOG_PREFIX} UKBinCollectionApp: {data}")
 
         if cm.expired:
-            _LOGGER.warning(f"{LOG_PREFIX} UKBinCollectionApp timeout expired during run")
-
+            _LOGGER.warning(
+                f"{LOG_PREFIX} UKBinCollectionApp timeout expired during run"
+            )
+
         return get_latest_collection_info(json.loads(data))
 
 
@@ -162,7 +168,9 @@ def apply_values(self):
             name = "{} {}".format(self.coordinator.name, self._bin_type)
         self._id = name
         self._name = name
-        self._next_collection = parser.parse(self.coordinator.data[self._bin_type], dayfirst=True).date()
+        self._next_collection = parser.parse(
+            self.coordinator.data[self._bin_type], dayfirst=True
+        ).date()
         self._hidden = False
         self._icon = "mdi:trash-can"
         self._colour = "red"
@@ -179,16 +187,22 @@ def apply_values(self):
         next_week_start = this_week_end + timedelta(days=1)
         next_week_end = next_week_start + timedelta(days=6)
 
-        self._days = (self._next_collection- now.date()).days
+        self._days = (self._next_collection - now.date()).days
         _LOGGER.info(f"{LOG_PREFIX} _days: {self._days}")
 
         if self._next_collection == now.date():
             self._state = "Today"
         elif self._next_collection == (now + timedelta(days=1)).date():
             self._state = "Tomorrow"
-        elif self._next_collection >= this_week_start and self._next_collection <= this_week_end:
+        elif (
+            self._next_collection >= this_week_start
+            and self._next_collection <= this_week_end
+        ):
             self._state = f"This Week: {self._next_collection.strftime('%A')}"
-        elif self._next_collection >= next_week_start and self._next_collection <= next_week_end:
+        elif (
+            self._next_collection >= next_week_start
+            and self._next_collection <= next_week_end
+        ):
             self._state = f"Next Week: {self._next_collection.strftime('%A')}"
         elif self._next_collection > next_week_end:
             self._state = f"Future: {self._next_collection}"
@@ -199,7 +213,6 @@ def apply_values(self):
 
         _LOGGER.info(f"{LOG_PREFIX} State of the sensor: {self._state}")
 
-
     @property
     def name(self):
         """Return the name of the bin."""
@@ -239,7 +252,7 @@ def colour(self):
     def unique_id(self):
         """Return a unique ID to use for this sensor."""
         return self._id
-    
+
     @property
     def bin_type(self):
         """Return the bin type."""

custom_components/uk_bin_collection/strings.json

@@ -19,6 +19,7 @@
                     "number": "House number of the address",
                     "usrn": "USRN (Unique Street Reference Number)",
                     "web_driver": "URL of the remote Selenium web driver to use",
+                    "headless": "Run Selenium in headless mode?",
                     "submit": "Submit"
                 },
                 "description": "Please refer to your councils [wiki](https://github.com/robbrad/UKBinCollectionData/wiki/Councils) entry for details on what to enter"
@@ -29,4 +30,4 @@
             "council": "Please select a council"
         }
     }
-}
+}

custom_components/uk_bin_collection/translations/en.json

@@ -19,6 +19,7 @@
                     "number": "House number of the address",
                     "usrn": "USRN (Unique Street Reference Number)",
                     "web_driver": "URL of the remote Selenium web driver to use",
+                    "headless": "Run Selenium in headless mode?",
                     "submit": "Submit"
                 },
                 "description": "Please refer to your councils [wiki](https://github.com/robbrad/UKBinCollectionData/wiki/Councils) entry for details on what to enter"
@@ -29,4 +30,4 @@
             "council": "Please select a council"
         }
     }
-}
+}

uk_bin_collection/tests/step_defs/step_helpers/file_handler.py

@@ -1,42 +1,38 @@
 import json
 import logging
-import os
 from jsonschema import validate, ValidationError
+from pathlib import Path
 
-logging.basicConfig(level=logging.INFO, format='%(levelname)s - %(message)s')
+logging.basicConfig(level=logging.INFO, format="%(levelname)s - %(message)s")
 
-def load_inputs_file(file_name):
-    cwd = os.getcwd()
-    with open(os.path.join(cwd, "uk_bin_collection", "tests", file_name)) as f:
-        data = json.load(f)
-        logging.info(f"{file_name} Input file loaded")
-    return data
+# Dynamically compute the base path relative to this file's location
+current_file_path = Path(__file__).resolve()
+BASE_PATH = current_file_path.parent.parent.parent.parent / "tests"
 
 
-def load_schema_file(file_name):
-    cwd = os.getcwd()
-    with open(os.path.join(cwd, "uk_bin_collection", "tests", file_name)) as f:
+def load_json_file(file_name):
+    file_path = BASE_PATH / file_name
+    with open(file_path) as f:
         data = json.load(f)
-        logging.info(f"{file_name} Schema file loaded")
+        logging.info(f"{file_name} file loaded")
     return data
 
 
 def validate_json(json_str):
     try:
-        json.loads(json_str)
+        return json.loads(json_str)
     except ValueError as err:
-        logging.info(f"The following error occured {err}")
-        return False
-    return True
+        logging.error(f"JSON validation error: {err}")
+        raise
 
 
 def validate_json_schema(json_str, schema):
-    json_data = json.loads(json_str)
+    json_data = validate_json(json_str)
     try:
         validate(instance=json_data, schema=schema)
     except ValidationError as err:
-        logging.info(f"The following error occured {err}")
+        logging.error(f"Schema validation error: {err}")
         logging.info(f"Data: {json_str}")
         logging.info(f"Schema: {schema}")
-        return False
+        raise
     return True

uk_bin_collection/tests/step_defs/test_validate_council.py

@@ -3,17 +3,32 @@
 import traceback
 from pytest_bdd import scenario, given, when, then, parsers
 from hamcrest import assert_that, equal_to
+from functools import wraps
 
 from step_helpers import file_handler
 from uk_bin_collection.uk_bin_collection import collect_data
 
-logging.basicConfig(level=logging.INFO, format='%(levelname)s - %(message)s')
+logging.basicConfig(level=logging.INFO, format="%(levelname)s - %(message)s")
+
 
 @scenario("../features/validate_council_outputs.feature", "Validate Council Output")
 def test_scenario_outline():
     pass
 
 
+def handle_test_errors(func):
+    @wraps(func)
+    def wrapper(*args, **kwargs):
+        try:
+            return func(*args, **kwargs)
+        except Exception as e:
+            logging.error(f"Error in test '{func.__name__}': {e}")
+            logging.error(traceback.format_exc())
+            raise e
+
+    return wrapper
+
+
 @pytest.fixture
 def context():
     class Context(object):
@@ -22,18 +37,20 @@ class Context(object):
     return Context()
 
 
+@handle_test_errors
 @given(parsers.parse("the council: {council_name}"))
 def get_council_step(context, council_name):
-    try:
-        council_input_data = file_handler.load_inputs_file("input.json")
-        context.metadata = council_input_data[council_name]
-    except Exception as err:
-        logging.error(traceback.format_exc())
-        logging.info(f"Validate Output: {err}")
-        raise (err)
+    council_input_data = file_handler.load_json_file("input.json")
+    context.metadata = council_input_data[council_name]
+
 
 # When we scrape the data from <council> using <selenium_mode> and the <selenium_url> is set.
-@when(parsers.parse("we scrape the data from {council} using {selenium_mode} and the {selenium_url} is set"))
+@handle_test_errors
+@when(
+    parsers.parse(
+        "we scrape the data from {council} using {selenium_mode} and the {selenium_url} is set"
+    )
+)
 def scrape_step(context, council, selenium_mode, selenium_url):
     context.council = council
     context.selenium_mode = selenium_mode
@@ -53,50 +70,35 @@ def scrape_step(context, council, selenium_mode, selenium_url):
     if "usrn" in context.metadata:
         usrn = context.metadata["usrn"]
         args.append(f"-us={usrn}")
+    if "headless" in context.metadata:
+        args.append(f"--headless")
     # TODO we should somehow run this test with and without this argument passed
     # TODO I do think this would make the testing of the councils a lot longer and cause a double hit from us
 
     # At the moment the feature file is set to local execution of the selenium so no url will be set
     # And it the behave test will execute locally
-    if selenium_mode != 'None' and selenium_url != 'None':
-        if selenium_mode != 'local': 
+    if selenium_mode != "None" and selenium_url != "None":
+        if selenium_mode != "local":
             web_driver = context.metadata["web_driver"]
             args.append(f"-w={web_driver}")
     if "skip_get_url" in context.metadata:
         args.append(f"-s")
 
-    try:
-        CollectData = collect_data.UKBinCollectionApp()
-        CollectData.set_args(args)
-        context.parse_result = CollectData.run()
-    except Exception as err:
-        logging.error(traceback.format_exc())
-        logging.info(f"Schema: {err}")
-        raise (err)
+    CollectData = collect_data.UKBinCollectionApp()
+    CollectData.set_args(args)
+    context.parse_result = CollectData.run()
 
 
+@handle_test_errors
 @then("the result is valid json")
 def validate_json_step(context):
-    try:
-        valid_json = file_handler.validate_json(context.parse_result)
-        assert_that(valid_json, True)
-    except Exception as err:
-        logging.error(traceback.format_exc())
-        logging.info(f"Validate Output: {err}")
-        logging.info(f"JSON Output: {context.parse_result}")
-        raise (err)
+    assert file_handler.validate_json(context.parse_result), "Invalid JSON output"
 
 
+@handle_test_errors
 @then("the output should validate against the schema")
 def validate_output_step(context):
-    try:
-        council_schema = file_handler.load_schema_file(f"output.schema")
-        schema_result = file_handler.validate_json_schema(
-            context.parse_result, council_schema
-        )
-        assert_that(schema_result, True)
-    except Exception as err:
-        logging.error(traceback.format_exc())
-        logging.info(f"Validate Output: {err}")
-        logging.info(f"JSON Output: {context.parse_result}")
-        raise (err)
+    council_schema = file_handler.load_json_file(f"output.schema")
+    assert file_handler.validate_json_schema(
+        context.parse_result, council_schema
+    ), "Schema validation failed"