mirror of
https://github.com/Akkudoktor-EOS/EOS.git
synced 2025-04-19 08:55:15 +00:00
d2ba0adb5f
* Add documentation to class_pv_forecast.py. Added documentation. Beware mostly generated by ChatGPT. Signed-off-by: Bobby Noelte <b0661n0e17e@gmail.com> * Add CacheFileStore, datetime and logger utilities. The `CacheFileStore` class is a singleton-based, thread-safe key-value store for managing temporary file objects, allowing the creation, retrieval, and management of cache files. The utility modules offer a flexible logging setup (`get_logger`) and utilities to handle different date-time formats (`to_datetime`, `to_timestamp`) and timezone detection (`to_timezone). - Cache files are automatically valid for the the current date unless specified otherwise. This is to mimic the current behaviour used in several classes. - The logger supports rotating log files to prevent excessive log file size. - The `to_datetime` and `to_timestamp`functions support a wide variety of input types and formats. They provide the time conversion that is e.g. used in PVForecast. Signed-off-by: Bobby Noelte <b0661n0e17e@gmail.com> * Improve testability of PVForecast Improvements for testing of PVForecast - Use common utility functions to allow for general testing at one spot. - to_datetime - CacheFileStore - Use logging instead of print to easily capture in testing. - Add validation of the json schema for Akkudoktor PV forecast data. - Allow to create an empty PVForecast instance as base instance for testing. - Make process_data() complete for filling a PVForecast instance for testing. - Normalize forecast datetime to timezone of system given in loaded data. - Do not print report but provide report for test checks. - Get rid of cache file path using the CachFileStore to automate cache file usage. - Improved module documentation. Signed-off-by: Bobby Noelte <b0661n0e17e@gmail.com> * Add test for PVForecast and newly extracted utility modules. - Add test for PVForecast - Add test for CacheFileStore in the new cachefilestore module - Add test for to_datetime, to_timestamp, to_timezone in the new datetimeutil module - Add test for get_logger in the new logutil module Signed-off-by: Bobby Noelte <b0661n0e17e@gmail.com> --------- Signed-off-by: Bobby Noelte <b0661n0e17e@gmail.com> Co-authored-by: Normann <github@koldrack.com>
107 lines
2.9 KiB
Python
107 lines
2.9 KiB
Python
import logging
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
|
|
import pytest
|
|
from xprocess import ProcessStarter
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def disable_debug_logging():
|
|
# Temporarily set logging level higher than DEBUG
|
|
logging.disable(logging.DEBUG)
|
|
yield
|
|
# Re-enable logging back to its original state after the test
|
|
logging.disable(logging.NOTSET)
|
|
|
|
|
|
def pytest_addoption(parser):
|
|
parser.addoption(
|
|
"--full-run", action="store_true", default=False, help="Run with all optimization tests."
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def is_full_run(request):
|
|
yield bool(request.config.getoption("--full-run"))
|
|
|
|
|
|
@pytest.fixture
|
|
def server(xprocess):
|
|
"""Fixture to start the server.
|
|
|
|
Provides URL of the server.
|
|
"""
|
|
|
|
class Starter(ProcessStarter):
|
|
# assure server to be installed
|
|
try:
|
|
subprocess.run(
|
|
[sys.executable, "-c", "import akkudoktoreosserver"],
|
|
check=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
)
|
|
except subprocess.CalledProcessError:
|
|
test_dir = os.path.dirname(os.path.realpath(__file__))
|
|
project_dir = os.path.abspath(os.path.join(test_dir, ".."))
|
|
subprocess.run(
|
|
[sys.executable, "-m", "pip", "install", "-e", project_dir],
|
|
check=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
)
|
|
|
|
# command to start server process
|
|
args = [sys.executable, "-m", "akkudoktoreosserver.flask_server"]
|
|
|
|
# startup pattern
|
|
pattern = "Debugger PIN:"
|
|
# search the first 12 lines for the startup pattern, if not found
|
|
# a RuntimeError will be raised informing the user
|
|
max_read_lines = 12
|
|
|
|
# will wait for 10 seconds before timing out
|
|
timeout = 30
|
|
|
|
# xprocess will now attempt to clean up upon interruptions
|
|
terminate_on_interrupt = True
|
|
|
|
# ensure process is running and return its logfile
|
|
logfile = xprocess.ensure("akkudoktoreosserver", Starter)
|
|
|
|
# create url/port info to the server
|
|
url = "http://127.0.0.1:8503"
|
|
yield url
|
|
|
|
# clean up whole process tree afterwards
|
|
xprocess.getinfo("akkudoktoreosserver").terminate()
|
|
|
|
|
|
@pytest.fixture
|
|
def other_timezone():
|
|
"""Fixture to temporarily change the timezone.
|
|
|
|
Restores the original timezone after the test.
|
|
"""
|
|
original_tz = os.environ.get("TZ", None)
|
|
|
|
other_tz = "Atlantic/Canary"
|
|
if original_tz == other_tz:
|
|
other_tz = "Asia/Singapore"
|
|
|
|
# Change the timezone to another
|
|
os.environ["TZ"] = other_tz
|
|
time.tzset() # For Unix/Linux to apply the timezone change
|
|
|
|
yield os.environ["TZ"] # Yield control back to the test case
|
|
|
|
# Restore the original timezone after the test
|
|
if original_tz:
|
|
os.environ["TZ"] = original_tz
|
|
else:
|
|
del os.environ["TZ"]
|
|
time.tzset() # Re-apply the original timezone
|