Merge pull request #82 from samirg1/add-tests-through-db

Add tests through db

.github/workflows/tests.yml

@@ -12,7 +12,7 @@ jobs:
         strategy:
             matrix:
                 os: [ubuntu-latest, windows-latest]
-                python-version: ["3.10", "3.11"]
+                python-version: ["3.12"]
 
         steps:
             - uses: actions/checkout@v3

README.md

@@ -1,14 +1,12 @@
 # AutoSMX
-Automation tool for SMX. 
+Automation tool for SMX.
 
 ![Tests](https://github.com/samirg1/ALTER-SMX-Tool/actions/workflows/tests.yml/badge.svg)
 
-Download the correct file for your OS below:
+Download below:
 - [Windows](https://github.com/samirg1/ALTER-SMX-Tool//releases/latest/download/AutoSMX.zip) (.exe)
     - Ensure to open and install the certificate in the zip `'code_sign.crt'`
-- [Mac](https://github.com/samirg1/ALTER-SMX-Tool//releases/latest/download/AutoSMX.dmg) (.dmg)
 
----
 ## License
 
 This project is licensed under the [Creative Commons Attribution-NonCommercial 4.0 International License](https://creativecommons.org/licenses/by-nc/4.0/).
@@ -21,7 +19,7 @@ Under the following terms:
 - Attribution — You must give appropriate credit, provide a link to the license, and indicate if changes were made. You may do so in any reasonable manner, but not in any way that suggests the licensor endorses you or your use.
 - Non-Commercial — You may not use the material for commercial purposes without obtaining prior written permission from the copyright holder.
 
-No additional restrictions 
+No additional restrictions
 — You may not apply legal terms or technological measures that legally restrict others from doing anything the license permits.
 
 For more details, please see the [full license terms](https://creativecommons.org/licenses/by-nc/4.0/legalcode).

requirements.txt

@@ -1,4 +1 @@
-pyautogui==0.9.54
-pyperclip==1.8.2
-pynput==1.7.6
-attrs==23.1.0
+attrs==23.1.0

requirements_dev.txt

@@ -1,9 +1,6 @@
 pytest==7.4.0
-mypy==1.5.0
+mypy==1.6.1
 tox==4.8.0
-black==22.10.0
+black==23.10.0
 pytest-cov==4.1.0
 attrs==23.1.0
-pyautogui==0.9.54
-pyperclip==1.8.2
-pynput==1.7.6

setup.cfg

@@ -4,15 +4,13 @@ description = GUI Automater for SMX
 author = Samir Gupta
 platforms = unix, linux, osx, cygwin, win32
 classifiers =
-    Programming Language :: Python :: 3.10
-    Programming Language :: Python :: 3.11
+    Programming Language :: Python :: 3.12
 
 [options]
 packages =
     design
     pages
     popups
-    gui
     storage
     db
 python_requires = >=3.10
@@ -21,7 +19,7 @@ package_dir =
 zip_safe = no
 
 [options.extras_require]
-testing = 
+testing =
     pytest>=7.0
     pytest-cov>=4.0
     mypy>=1.0
@@ -32,6 +30,5 @@ testing =
 design = py.typed
 pages = py.typed
 popups = py.typed
-gui = py.typed
 storage = py.typed
-db = py.typed
+db = py.typed

src/App.py

@@ -4,15 +4,13 @@
 import tkinter
 from tkinter import PhotoImage, ttk
 
-from design.TestJobManager import TestJobManager
-from pages.CalibrationPage import CalibrationPage
-from pages.JobPage import JobPage
+from design.JobManager import JobManager
+from pages.ProblemPage import ProblemPage
 from pages.Page import TPAGES, Page, SharedPageInfo
 from pages.SettingsPage import SettingsPage
 from pages.TestPage import TestPage
 from pages.TutorialPage import TutorialPage
 from storage.Storage import Storage
-from db.set_favourites import set_favourites
 
 _APPLICATION_PATH = os.path.dirname(sys.executable)
 
@@ -31,12 +29,11 @@ def __init__(self) -> None:
         self.columnconfigure(0, weight=1)
         self.rowconfigure(0, weight=1)
 
-        shared = SharedPageInfo({}, TestJobManager(), Storage(pathlib.Path(_APPLICATION_PATH, "store.json")))
+        shared = SharedPageInfo({}, JobManager(), Storage(pathlib.Path(_APPLICATION_PATH, "store.json")))
         self.pages: dict[TPAGES, Page] = {
             "TUTORIAL": TutorialPage(self._frame(), self.change_page, shared),
-            "CALIBRATION": CalibrationPage(self._frame(), self.change_page, shared),
             "SETTINGS": SettingsPage(self._frame(), self.change_page, shared),
-            "JOB": JobPage(self._frame(), self.change_page, shared),
+            "PROBLEM": ProblemPage(self._frame(), self.change_page, shared),
             "TEST": TestPage(self._frame(), self.change_page, shared),
         }
         self.current_page: Page | None = None
@@ -63,7 +60,6 @@ def change_page(self, page: TPAGES) -> None:
 
 
 def main() -> None:
-    set_favourites()
     App().mainloop()
 
 

src/db/add_test.py

@@ -0,0 +1,56 @@
+from datetime import datetime, timedelta
+
+from db.get_connection import DatabaseFilenames, get_connection
+from db.models import JobModel, ScriptLineModel, ScriptTesterModel, TestModel
+from design.Problem import Problem
+from design.Script import ScriptLine
+from design.Test import Test
+
+
+def add_test(test: Test, problem: Problem) -> None:
+    next_spt_date = (datetime.now() + timedelta(days=366)).strftime(r"%Y-%m-%d %H:%M:%S.%f")[:-3]
+
+    with get_connection(DatabaseFilenames.TESTS, mode="rw") as connection, get_connection(DatabaseFilenames.ASSETS, mode="rw") as asset_connection:
+        with connection, asset_connection:
+            TestModel(test, problem).insert(connection)
+            ScriptTesterModel(test).insert(connection)
+
+            lines = test.script.lines
+            if test.script.number == 1287:
+                track_header_line = ScriptLine(text="Disclaimer: Tests carried out are subject to conditions, available on request", number=1)
+                lines = (track_header_line, *lines)
+
+            for line in lines:
+                ScriptLineModel(test, line).insert(connection)
+
+            for job in test.jobs:
+                JobModel(test, problem, job).insert(connection)
+
+            asset_connection.execute(
+                """
+                UPDATE DEVICEA4
+                SET service_last = ?, service_next = ?
+                WHERE logical_name = ? AND service_type = ?;
+                """,
+                (test.date, next_spt_date, test.item.number, test.script.service_type),
+            )
+
+            services: list[tuple[str, float, str, str]] = asset_connection.execute(
+                """
+                SELECT service_type, service_interval, service_last, service_next
+                FROM DEVICEA4
+                WHERE logical_name = ?;
+                """,
+                (test.item.number,),
+            ).fetchall()
+
+            servicearray = "\n".join("^".join(f"{int(s) if isinstance(s, float) else s}" for s in service) + "^" for service in services)
+
+            connection.execute(
+                """
+                UPDATE devicem1_PS
+                SET last_spt_date = ?, next_spt_date = ?, servicearray = ?
+                WHERE logical_name = ?;
+                """,
+                (test.date, next_spt_date, servicearray, test.item.number),
+            )

src/db/convert_stringed_date.py

@@ -1,6 +1,8 @@
 from datetime import datetime
 
 
-def convert_stringed_date(date_string: str) -> datetime:
-    """Converts date from 023-10-05 15:12:23.260 format to datetime object"""
+def convert_stringed_date(date_string: str | None) -> datetime | None:
+    """Converts date from 2023-10-05 15:12:23.260 format to datetime object"""
+    if date_string is None:
+        return None
     return datetime.strptime(date_string[:-4], r"%Y-%m-%d %H:%M:%S")

src/db/edit_test.py

@@ -0,0 +1,16 @@
+from db.get_connection import get_connection, DatabaseFilenames
+from db.add_test import add_test
+from design.Problem import Problem
+from design.Test import Test
+
+
+def edit_test(test: Test, problem: Problem, *, remove_only: bool = False) -> None:
+    with get_connection(DatabaseFilenames.TESTS, mode="rw") as connection:
+        with connection:
+            connection.execute("DELETE FROM SCMobileTestsm1 WHERE test_id = ?;", (test.id,))
+            connection.execute("DELETE FROM SCMobileTesterNumbersm1 WHERE test_id = ?", (test.id,))
+            connection.execute("DELETE FROM SCMobileTestLinesm1 WHERE test_id = ?", (test.id,))
+            connection.execute("DELETE FROM SCMProbsUploadm1 WHERE test_id = ?", (test.id,))
+
+    if not remove_only:
+        add_test(test, problem)

src/db/get_connection.py

@@ -1,14 +1,21 @@
 import os
+import sqlite3
 from contextlib import contextmanager
+from enum import StrEnum
 from typing import Generator, Literal
-import sqlite3
+
+BASE_FILE = rf"C:\Users\{os.getenv('USERNAME')}\AppData\Local\SMXMobile"
 
 
-_BASE_FILE = rf"C:\Users\{os.getenv('USERNAME')}\AppData\Local\SMXMobile"
+class DatabaseFilenames(StrEnum):
+    TESTS = "SCMTests"
+    ASSETS = "SCMAssets"
+    LOOKUP = "SCMLookup"
+    SETTINGS = "Settings"
 
 
 @contextmanager
-def get_connection(filename: str, mode: Literal["ro", "rw"] = "ro") -> Generator[sqlite3.Connection, None, None]:
-    connection = sqlite3.connect(rf"file:{_BASE_FILE}\{filename}.sdb?mode={mode}", uri=True)
+def get_connection(filename: DatabaseFilenames, *, mode: Literal["ro", "rw"] = "ro") -> Generator[sqlite3.Connection, None, None]:
+    connection = sqlite3.connect(rf"file:{BASE_FILE}\{filename}.sdb?mode={mode}", uri=True)
     yield connection
     connection.close()

src/db/get_items.py

@@ -0,0 +1,16 @@
+from db.get_connection import DatabaseFilenames, get_connection
+from design.Item import Item
+
+
+def get_items(item_number: str) -> list[Item]:
+    with get_connection(DatabaseFilenames.TESTS) as connection:
+        item_fields = connection.execute(
+            """
+            SELECT logical_name, customer_barcode, description, model, manufacturer, serial_no_, room, last_spt_date
+            FROM devicem1_PS
+            WHERE logical_name LIKE ?;
+            """,
+            (item_number + "%",),
+        ).fetchall()
+
+    return [Item(*fields) for fields in item_fields]

src/db/get_new_test_id.py

@@ -0,0 +1,33 @@
+from db.get_connection import DatabaseFilenames, get_connection
+
+
+class NoTestIDsError(RuntimeError):
+    ...
+
+
+def get_new_test_id() -> str:
+    with get_connection(DatabaseFilenames.SETTINGS, mode="rw") as connection:
+        res: tuple[str, int, int] | None = connection.execute(
+            """
+            SELECT TABLENAME, LASTUSED, LASTRESERVED
+            FROM SCMIDTABLE
+            WHERE TABLENAME == 'SCMobileTestsm1' AND LASTUSED <> LASTRESERVED;
+            """
+        ).fetchone()
+
+        if res is None:
+            raise NoTestIDsError
+
+        _, current, end = res
+
+        with connection:
+            connection.execute(
+                """
+                UPDATE SCMIDTABLE
+                SET LASTUSED = ?
+                WHERE TABLENAME == 'SCMobileTestsm1' AND LASTRESERVED = ?;
+                """,
+                (current + 1, end),
+            )
+
+    return f"SMX{str(current+1).zfill(10)}"

src/db/get_open_problems.py

@@ -1,22 +1,24 @@
-from db.get_connection import get_connection
-from db.convert_stringed_date import convert_stringed_date
 from typing import NamedTuple
 
+from db.convert_stringed_date import convert_stringed_date
+from db.get_connection import DatabaseFilenames, get_connection
+
 
-class Problem(NamedTuple):
+class OpenProblem(NamedTuple):
     number: str
     description: str
     date_opened: str
     asset_description: str
     asset_serial: str
 
     def __repr__(self) -> str:
-        date_opened = convert_stringed_date(self.date_opened).strftime(r"%d-%m-%Y")
-        return f"{self.number} - {date_opened}\n{self.description}\n{self.asset_description} ({self.asset_serial})"
+        converted = convert_stringed_date(self.date_opened)
+        date_opened = "Not found" if converted is None else converted.strftime(r"%d-%m-%Y")
+        return f"{self.number} - Opened: {date_opened}\n{self.description}\n{self.asset_description} ({self.asset_serial})"
 
 
-def get_open_problems(location: str) -> list[Problem]:
-    with get_connection("SCMLookup") as connection:
+def get_open_problems(location: str) -> list[OpenProblem]:
+    with get_connection(DatabaseFilenames.LOOKUP) as connection:
         results = connection.execute(
             """
             SELECT number, brief_description, open_time, asset_description, serial_no_
@@ -26,4 +28,4 @@ def get_open_problems(location: str) -> list[Problem]:
             (location,),
         ).fetchall()
 
-    return [Problem(*res) for res in results]
+    return [OpenProblem(*res) for res in results]

src/db/get_overall_results.py

@@ -1,6 +1,7 @@
-from typing import NamedTuple
-from db.get_connection import get_connection
 import functools
+from typing import NamedTuple
+
+from db.get_connection import DatabaseFilenames, get_connection
 
 
 class TestResult(NamedTuple):
@@ -10,7 +11,7 @@ class TestResult(NamedTuple):
 
 @functools.lru_cache(maxsize=5)
 def get_overall_results(customer_id: int) -> list[TestResult]:
-    with get_connection("SCMLookup") as connection:
+    with get_connection(DatabaseFilenames.LOOKUP) as connection:
         results: list[tuple[str, str]] = connection.execute(
             """
             SELECT overall_id, overall_text

src/db/get_problems.py

@@ -0,0 +1,16 @@
+from db.get_connection import DatabaseFilenames, get_connection
+from design.Problem import Problem
+
+
+def get_problems(problem_number: str) -> list[Problem]:
+    with get_connection(DatabaseFilenames.LOOKUP) as connection:
+        problem_fields = connection.execute(
+            """
+            SELECT company, location, dept, number, customer_no_
+            FROM probsummarym1
+            WHERE number LIKE ? OR number LIKE ?;
+            """,
+            (f"PM{problem_number}%", f"{problem_number}%"),
+        ).fetchall()
+
+    return [Problem(*fields) for fields in problem_fields]