Fix failing test

.github/workflows/tests.yaml

@@ -11,7 +11,7 @@ jobs:
       fail-fast: false
       matrix:
         python-version: ["3.11"]
-    
+
     # Service containers to run with `container-job`
     services:
       # Label used to access the service container
@@ -32,27 +32,27 @@ jobs:
           --health-timeout 5s
           --health-retries 5
     steps:
-    - run: echo "🎉 The job was automatically triggered by a ${{ github.event_name }} event."
-    - run: echo "🐧 This job is now running on a ${{ runner.os }} server hosted by GitHub!"
-    - run: echo "🔎 The name of your branch is ${{ github.ref }} and your repository is ${{ github.repository }}."
-    - uses: actions/checkout@v3
-    - name: Set up Python ${{ matrix.python-version }}
-      uses: actions/setup-python@v3
-      with:
-        python-version: ${{ matrix.python-version }}
-    - name: Install dependencies
-      run: |
-        python -m pip install --upgrade pip
-        python -m pip install flake8 pytest
-        if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
-    - name: Lint with flake8
-      run: |
-        # stop the build if there are Python syntax errors or undefined names
-        flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
-    - name: Test with pytest
-      run: |
-        pytest tests
-      env:
+      - run: echo "🎉 The job was automatically triggered by a ${{ github.event_name }} event."
+      - run: echo "🐧 This job is now running on a ${{ runner.os }} server hosted by GitHub!"
+      - run: echo "🔎 The name of your branch is ${{ github.ref }} and your repository is ${{ github.repository }}."
+      - uses: actions/checkout@v3
+      - name: Set up Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v3
+        with:
+          python-version: ${{ matrix.python-version }}
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          python -m pip install flake8 pytest
+          if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
+      - name: Lint with flake8
+        run: |
+          # stop the build if there are Python syntax errors or undefined names
+          flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
+      - name: Test with pytest
+        run: |
+          pytest tests
+        env:
           # Env vars for pytest
           POSTGRES_USERNAME: postgres
           POSTGRES_HOST: localhost
@@ -61,4 +61,4 @@ jobs:
           POSTGRES_PLAYBACK_DB_NAME: postgres_playback_test
           POSTGRES_PORT: 5432
           DQM_CR_USERNAMES: "user:password"
-    - run: echo "🍏 This job's status is ${{ job.status }}."
+      - run: echo "🍏 This job's status is ${{ job.status }}."

db.py

@@ -1,4 +1,6 @@
 ### DQM^2 Mirror DB === >
+import os
+import sys
 import psycopg2
 import sqlalchemy
 from sqlalchemy_utils import database_exists
@@ -7,6 +9,10 @@
 from datetime import datetime
 from exceptions import DatabaseNotFoundError
 
+sys.path.append(os.path.join(os.path.dirname(__file__), "."))
+
+from dqmsquare_cfg import TZ
+
 
 class DQM2MirrorDB:
     """
@@ -165,10 +171,12 @@ def fill_graph(self, header: dict, document: dict) -> int:
 
         rev = header.get("_rev", -1)
         timestamp = extra.get(
-            "timestamp", datetime(2012, 3, 3, 10, 10, 10, 0).timestamp()
+            "timestamp",
+            TZ.localize(datetime(2012, 3, 3, 10, 10, 10, 0), is_dst=None).timestamp(),
         )
         global_start = extra.get(
-            "global_start", datetime(2012, 3, 3, 10, 10, 10, 0).timestamp()
+            "global_start",
+            TZ.localize(datetime(2012, 3, 3, 10, 10, 10, 0), is_dst=None).timestamp(),
         )
 
         stream_data = str(extra.get("streams", ""))
@@ -224,8 +232,8 @@ def get_graphs_data(self, run) -> list:
         answer = list(answer[0])
         if answer[-2]:
             answer[-2] = eval(answer[-2])  # TODO: Not secure!!!!!!
-        answer[3] = answer[3].timestamp()
-        answer[4] = answer[4].timestamp()
+        answer[3] = TZ.localize(answer[3], is_dst=None).timestamp()
+        answer[4] = TZ.localize(answer[4], is_dst=None).timestamp()
 
         return answer
 
@@ -253,7 +261,8 @@ def fill(self, header: dict, document: dict) -> int:
             pass
         fi_state = document.get("fi_state", "")
         timestamp = header.get(
-            "timestamp", datetime(2012, 3, 3, 10, 10, 10, 0).timestamp()
+            "timestamp",
+            TZ.localize(datetime(2012, 3, 3, 10, 10, 10, 0), is_dst=None).timestamp(),
         )
         try:
             timestamp = datetime.fromtimestamp(timestamp)

dqmsquare_cfg.py

@@ -6,9 +6,15 @@
 """
 
 import os
+import pytz
 import tempfile
 from dotenv import load_dotenv
 
+# Important for converting datetime objects (from the database)
+# to timestamps. Github actions, for example, run in different timezones,
+# leading to different timestamps and failing tests.
+TZ = pytz.timezone("Europe/Zurich")
+
 
 def format_db_uri(
     username: str = "postgres",

grabber.py

@@ -14,7 +14,7 @@
 
 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
 
-log = logging.getLogger(__name__)
+logger = logging.getLogger(__name__)
 cfg = dqmsquare_cfg.load_cfg()
 
 
@@ -34,7 +34,7 @@ def get_documents_from_fff(host: str, port: int = cfg["FFF_PORT"], runs_ids: lis
 
     jsn = {"event": "request_documents", "ids": runs_ids}
     data = json.dumps({"messages": [json.dumps(jsn)]})
-    log.debug(f"POSTing to '{url}' with data: {jsn}")
+    logger.debug(f"POSTing to '{url}' with data: {jsn}")
     r = requests.post(
         url,
         data=data,
@@ -44,7 +44,7 @@ def get_documents_from_fff(host: str, port: int = cfg["FFF_PORT"], runs_ids: lis
         cookies=cookies,
         timeout=30,
     )
-    log.debug(f"Got {len(r.content)} byte response.")
+    logger.debug(f"Got {len(r.content)} byte response.")
 
     return r.content
 
@@ -69,7 +69,7 @@ def get_headers_from_fff(host: str, port: int = cfg["FFF_PORT"], revision: int =
 
     jsn = {"event": "sync_request", "known_rev": str(revision)}
     data = json.dumps({"messages": [json.dumps(jsn)]})
-    log.debug(f"POSTing to '{url}' with data: {jsn}")
+    logger.debug(f"POSTing to '{url}' with data: {jsn}")
     r = requests.post(
         url,
         data=data,
@@ -79,7 +79,7 @@ def get_headers_from_fff(host: str, port: int = cfg["FFF_PORT"], revision: int =
         cookies=cookies,
         timeout=30,
     )
-    log.debug(f"Got response of length {len(r.content)}.")
+    logger.debug(f"Got response of length {len(r.content)}.")
 
     return r.content
 
@@ -97,7 +97,7 @@ def get_latest_info_from_host(host: str, rev: int, db: DQM2MirrorDB) -> None:
     database.
     """
     # global bad_rvs
-    log.info(f"Updating host {host}, starting from revision {str(rev)}")
+    logger.info(f"Updating host {host}, starting from revision {str(rev)}")
     if not rev:
         rev = 0
 
@@ -106,30 +106,30 @@ def get_latest_info_from_host(host: str, rev: int, db: DQM2MirrorDB) -> None:
         headers_answer = json.loads(json.loads(headers_answer)["messages"][0])
         headers = headers_answer["headers"]
         rev = headers_answer["rev"]
-        log.debug(
+        logger.debug(
             f"Got {headers_answer['total_sent']} headers, from {rev[0]} to {rev[1]}."
         )
     except Exception as e:
-        log.warning(f"Error when getting headers: {repr(e)}")
-        log.warning(f"Got response: {headers_answer}")
-        log.warning(traceback.format_exc())
+        logger.warning(f"Error when getting headers: {repr(e)}")
+        logger.warning(f"Got response: {headers_answer}")
+        logger.warning(traceback.format_exc())
         return
 
     if not len(headers):
         return
     for i, header in enumerate(headers):
         id = header["_id"]
-        log.info(f"Processing header {str(id)} ({i+1}/{len(headers)})")
+        logger.info(f"Processing header {str(id)} ({i+1}/{len(headers)})")
 
         is_bu = host[0] == "b"
         if is_bu and "analyze_files" not in id:
-            log.debug("Skip, no 'analyze_files' key")
+            logger.debug("Skip, no 'analyze_files' key")
             continue
         document_answer = get_documents_from_fff(host, runs_ids=[id])
         document = json.loads(json.loads(document_answer)["messages"][0])["documents"][
             0
         ]
-        log.debug("Filling info into DB ... ")
+        logger.debug("Filling info into DB ... ")
 
         # BU sends us file delivery graph info, FUs sends us logs and event processing rates.
         answer = db.fill_graph(header, document) if is_bu else db.fill(header, document)
@@ -144,7 +144,7 @@ def get_latest_info_from_hosts(hosts: list[str], db: DQM2MirrorDB) -> None:
     in hostnames, storing the new information into db.
     """
     for host in hosts:
-        log.debug(f"Getting latest rev for {host} from DB.")
+        logger.debug(f"Getting latest rev for {host} from DB.")
         rev = db.get_latest_revision(host)
         get_latest_info_from_host(host=host, rev=rev, db=db)
 
@@ -156,7 +156,7 @@ def get_latest_info_from_hosts(hosts: list[str], db: DQM2MirrorDB) -> None:
 
     if len(sys.argv) > 1 and sys.argv[1] == "playback":
         set_log_handler(
-            log,
+            logger,
             cfg["ROBBER_LOG_PATH_PLAYBACK"],
             cfg["LOGGER_ROTATION_TIME"],
             cfg["LOGGER_MAX_N_LOG_FILES"],
@@ -165,7 +165,7 @@ def get_latest_info_from_hosts(hosts: list[str], db: DQM2MirrorDB) -> None:
         run_modes = ["playback"]
     elif len(sys.argv) > 1 and sys.argv[1] == "production":
         set_log_handler(
-            log,
+            logger,
             cfg["ROBBER_LOG_PATH_PRODUCTION"],
             cfg["LOGGER_ROTATION_TIME"],
             cfg["LOGGER_MAX_N_LOG_FILES"],
@@ -174,7 +174,7 @@ def get_latest_info_from_hosts(hosts: list[str], db: DQM2MirrorDB) -> None:
         run_modes = ["production"]
     else:
         set_log_handler(
-            log,
+            logger,
             cfg["GRABBER_LOG_PATH"],
             cfg["LOGGER_ROTATION_TIME"],
             cfg["LOGGER_MAX_N_LOG_FILES"],
@@ -186,8 +186,8 @@ def get_latest_info_from_hosts(hosts: list[str], db: DQM2MirrorDB) -> None:
         handler2.setFormatter(formatter)
         level = logging.DEBUG if cfg["GRABBER_DEBUG"] else logging.INFO
         handler2.setLevel(level=level)
-        log.addHandler(handler2)
-        log.info(f"Configured logger for grabber, level={level}")
+        logger.addHandler(handler2)
+        logger.info(f"Configured logger for grabber, level={level}")
 
     ### global variables and auth cookies
     cmsweb_proxy_url = cfg["CMSWEB_FRONTEND_PROXY_URL"]
@@ -196,21 +196,21 @@ def get_latest_info_from_hosts(hosts: list[str], db: DQM2MirrorDB) -> None:
     env_secret = os.environ.get("DQM_FFF_SECRET")
     if env_secret:
         fff_secret = env_secret
-        log.debug("Found secret in environmental variables")
+        logger.debug("Found secret in environmental variables")
     else:
-        log.warning("No secret found in environmental variables")
+        logger.warning("No secret found in environmental variables")
 
     # Trailing whitespace in secret leads to crashes, strip it
     cookies = {str(cfg["FFF_SECRET_NAME"]): env_secret.strip()}
 
     # DB CONNECTION
     db_playback, db_production = None, None
     if "playback" in run_modes:
-        db_playback = DQM2MirrorDB(log, cfg["DB_PLAYBACK_URI"])
+        db_playback = DQM2MirrorDB(logger, cfg["DB_PLAYBACK_URI"])
     if "production" in run_modes:
-        db_production = DQM2MirrorDB(log, cfg["DB_PRODUCTION_URI"])
+        db_production = DQM2MirrorDB(logger, cfg["DB_PRODUCTION_URI"])
 
-    log.info("Starting loop for modes " + str(run_modes))
+    logger.info("Starting loop for modes " + str(run_modes))
 
     # Main Loop.
     # Fetches data for CMSSW jobs from DQM^2, and stores it into the database.
@@ -232,12 +232,12 @@ def get_latest_info_from_hosts(hosts: list[str], db: DQM2MirrorDB) -> None:
         except KeyboardInterrupt:
             break
         except Exception as error:
-            log.warning(f"Crashed in loop with error: {repr(error)}")
-            log.warning(f"Traceback: {traceback.format_exc()}")
+            logger.warning(f"Crashed in loop with error: {repr(error)}")
+            logger.warning(f"Traceback: {traceback.format_exc()}")
             continue
 
         if cfg["ENV"] != "development":
-            log.debug("z-Z-z until next iteration")
+            logger.debug("z-Z-z until next iteration")
             time.sleep(int(cfg["SLEEP_TIME"]))
 
         # if len(bad_rvs):

requirements.txt

@@ -1,9 +1,10 @@
 requests
 cernrequests
-Flask<3.0.0
-SQLAlchemy<2.0.0    
+Flask<3.0
+SQLAlchemy<2.0   
 sqlalchemy-utils
 gunicorn<21.0.0
 psycopg2-binary
 werkzeug
-python-dotenv<2.0.0
+python-dotenv<2.0.0
+pytz

tests/test_1.py

@@ -12,7 +12,7 @@
 from sqlalchemy import create_engine, insert
 from sqlalchemy_utils import create_database, database_exists, drop_database
 from custom_logger import dummy_log
-from dqmsquare_cfg import format_db_uri, load_cfg
+from dqmsquare_cfg import format_db_uri, TZ
 
 
 def format_entry_to_db_entry(graph_entry: list, datetime_cols: list):
@@ -226,6 +226,7 @@ def test_db_9(testing_database):
     print("Check DQM2MirrorDB.get_graphs_data()")
     truth = TEST_DB_9_TRUTH
     answer = json.dumps(testing_database.get_graphs_data(358792))
+    print(answer)
     assert truth == answer
 
 
@@ -235,8 +236,8 @@ def test_db_10(testing_database):
         123456,
         -1,
         "id",
-        datetime(2012, 3, 3, 10, 10, 10).isoformat(),
-        datetime(2012, 3, 3, 10, 10, 10).isoformat(),
+        TZ.localize(datetime(2012, 3, 3, 10, 10, 10), is_dst=None).timestamp(),
+        TZ.localize(datetime(2012, 3, 3, 10, 10, 10), is_dst=None).timestamp(),
         "",
         "",
     ]