From c5308e1920bc49d3401ec2b7e8e9897fc269c401 Mon Sep 17 00:00:00 2001 From: "i.alexandrov" Date: Tue, 9 Dec 2025 14:45:07 +0300 Subject: [PATCH 01/25] [examples] Add API key integration example --- examples/stand_cloud_integration/README.md | 3 + examples/stand_cloud_integration/auth_flow.py | 36 ++++ .../stand_cloud_thirdparty_auth/README.md | 3 - .../stand_cloud_thirdparty_auth/auth_flow.py | 79 -------- .../token_update.py | 190 ------------------ 5 files changed, 39 insertions(+), 272 deletions(-) create mode 100644 examples/stand_cloud_integration/README.md create mode 100644 examples/stand_cloud_integration/auth_flow.py delete mode 100644 examples/stand_cloud_thirdparty_auth/README.md delete mode 100644 examples/stand_cloud_thirdparty_auth/auth_flow.py delete mode 100644 examples/stand_cloud_thirdparty_auth/token_update.py diff --git a/examples/stand_cloud_integration/README.md b/examples/stand_cloud_integration/README.md new file mode 100644 index 00000000..ca9e88dd --- /dev/null +++ b/examples/stand_cloud_integration/README.md @@ -0,0 +1,3 @@ +# StandCloud third party integration + +Example documentation: https://everypinio.github.io/hardpy/examples/stand_cloud_thirdparty_integration/ diff --git a/examples/stand_cloud_integration/auth_flow.py b/examples/stand_cloud_integration/auth_flow.py new file mode 100644 index 00000000..135ac2b5 --- /dev/null +++ b/examples/stand_cloud_integration/auth_flow.py @@ -0,0 +1,36 @@ +import sys + +import requests + +######################################## +# Change API_KEY to your api key +API_KEY = "your_api_key" +######################################## + +SSL_VERIFY = True +BASE_URL = "https://standcloud.io" +API_URL = BASE_URL + "/integration/api/v1" # API endpoints + +HEALTHCHECK_URL = API_URL + "/healthcheck" +response = requests.get(HEALTHCHECK_URL, verify=SSL_VERIFY) + +if response.status_code != 200: + print(response.text) + sys.exit(1) +print("StandCloud is up and running") + +# Test API call +header = { + "Authorization": f"Bearer {API_KEY}", + "Content-type": "application/json", + "Accept": "text/plain", +} +print("\nheader: ", header) + +USER_INFO_URL = API_URL + "/test_run" +response = requests.get(USER_INFO_URL, headers=header, verify=SSL_VERIFY) + +if response.status_code != 200: + print(response.text) + sys.exit(1) +print("\nTest run list: ", response.text) diff --git a/examples/stand_cloud_thirdparty_auth/README.md b/examples/stand_cloud_thirdparty_auth/README.md deleted file mode 100644 index 9f8491e4..00000000 --- a/examples/stand_cloud_thirdparty_auth/README.md +++ /dev/null @@ -1,3 +0,0 @@ -# StandCloud third party authorization - -Example documentation: https://everypinio.github.io/hardpy/examples/stand_cloud_thirdparty_auth/ diff --git a/examples/stand_cloud_thirdparty_auth/auth_flow.py b/examples/stand_cloud_thirdparty_auth/auth_flow.py deleted file mode 100644 index 4a3e2eea..00000000 --- a/examples/stand_cloud_thirdparty_auth/auth_flow.py +++ /dev/null @@ -1,79 +0,0 @@ -# This script demonstrates the StandCloud authentication process. -# Change the URL to your StandCloud address, nothing else needs to be changed. - -import json -import sys -import time - -import requests - -######################################## -# Change URL to your StandCloud address -BASE_URL = "https://demo.standcloud.io" -######################################## - -# Confugirable parameters, False if the server is self-signed -SSL_VERIFY = True - -# OAuth client configuration -# client_id must be "hardpy-report-uploader" -CLIENT_ID = "hardpy-report-uploader" - -# API endpoints -API_URL = BASE_URL + "/hardpy/api/v1" -DEVICE_AUTHORIZATION_URL = BASE_URL + "/auth/api/oidc/device-authorization" -TOKEN_URL = BASE_URL + "/auth/api/oidc/token" - - -# Device authorization aequest -data = { - "client_id": CLIENT_ID, - "scope": "offline_access authelia.bearer.authz", - "audience": API_URL, -} - -req = requests.post(DEVICE_AUTHORIZATION_URL, data=data, verify=SSL_VERIFY, timeout=10) -response = json.loads(req.content) -verification_uri_complete = response["verification_uri_complete"] -interval = response["interval"] - -# Token request -if "error" in response: - error = response["error"] - error_description = response["error_description"] - print(f"{error}: {error_description}") - sys.exit(1) - -print(f"\nVerification URI: {verification_uri_complete}") - -data = { - "client_id": CLIENT_ID, - "grant_type": "urn:ietf:params:oauth:grant-type:device_code", - "device_code": response["device_code"], -} - -# Token response -while True: - req = requests.post(TOKEN_URL, data=data, verify=SSL_VERIFY, timeout=10) - response = json.loads(req.content) - print(".", end="") - if "access_token" in response: - print("\nToken info:", response) - break - time.sleep(interval) - -# Test API call -header = { - "Authorization": "Bearer {}".format(response["access_token"]), - "Content-type": "application/json", - "Accept": "text/plain", -} -print("\nheader: ", header) - -USER_INFO_URL = API_URL + "/healthcheck" -response = requests.get(USER_INFO_URL, headers=header, verify=SSL_VERIFY, timeout=10) - -if response.status_code != 200: - print(response.text) - sys.exit(1) -print("\nOAuth Authenticated and Authorized API access") diff --git a/examples/stand_cloud_thirdparty_auth/token_update.py b/examples/stand_cloud_thirdparty_auth/token_update.py deleted file mode 100644 index 7eb96031..00000000 --- a/examples/stand_cloud_thirdparty_auth/token_update.py +++ /dev/null @@ -1,190 +0,0 @@ -# This script demonstrates the StandCloud authentication process with -# token updating. Don't use the save as file option, it's insecure. -# Change the URL to your StandCloud address, nothing else needs to be changed. - -import json -import sys -import time -from pathlib import Path - -import requests - -######################################## -# Change URL to your StandCloud address -BASE_URL = "https://demo.standcloud.io" -######################################## - -# Configurable parameters, False if the server is self-signed -SSL_VERIFY = True - -# OAuth client configuration -# client_id must be "hardpy-report-uploader" -CLIENT_ID = "hardpy-report-uploader" - -# API endpoints -API_URL = BASE_URL + "/hardpy/api/v1" -DEVICE_AUTHORIZATION_URL = BASE_URL + "/auth/api/oidc/device-authorization" -TOKEN_URL = BASE_URL + "/auth/api/oidc/token" -HEALTHCHECK_URL = API_URL + "/healthcheck" -TOKEN_FILE = Path(__file__).parent / "token_info.json" - - -def authenticate(): - data = { - "client_id": CLIENT_ID, - "scope": "offline_access authelia.bearer.authz", - "audience": API_URL, - } - - req = requests.post( - DEVICE_AUTHORIZATION_URL, data=data, verify=SSL_VERIFY, timeout=10 - ) - response = req.json() - - if "error" in response: - error = response["error"] - error_description = response["error_description"] - print(f"{error}: {error_description}") - sys.exit(1) - - verification_uri_complete = response["verification_uri_complete"] - interval = response["interval"] - - print(f"\nVerification URI: {verification_uri_complete}") - - data = { - "client_id": CLIENT_ID, - "grant_type": "urn:ietf:params:oauth:grant-type:device_code", - "device_code": response["device_code"], - } - - while True: - req = requests.post(TOKEN_URL, data=data, verify=SSL_VERIFY, timeout=10) - response = req.json() - if "access_token" in response: - print("\nAuthentication successful") - break - time.sleep(interval) - - # Calculate expiration time - expires_at = time.time() + response["expires_in"] - token_info = { - "access_token": response["access_token"], - "refresh_token": response["refresh_token"], - "expires_at": expires_at, - } - return token_info - - -def refresh_tokens(refresh_token_value): - data = { - "client_id": CLIENT_ID, - "grant_type": "refresh_token", - "refresh_token": refresh_token_value, - } - - try: - req = requests.post(TOKEN_URL, data=data, verify=SSL_VERIFY, timeout=10) - response = req.json() - - if "access_token" not in response: - print( - "Token refresh failed. Error:", response.get("error", "unknown error") - ) - return None - - expires_at = time.time() + response["expires_in"] - token_info = { - "access_token": response["access_token"], - "expires_at": expires_at, - } - - # Update refresh token if new one is provided - if "refresh_token" in response: - token_info["refresh_token"] = response["refresh_token"] - print("Access and refresh tokens updated") - else: - print("Access token updated") - - return token_info - - except Exception as e: - print("Token refresh failed:", str(e)) - return None - - -def healthcheck(access_token): - headers = { - "Authorization": f"Bearer {access_token}", - "Content-type": "application/json", - "Accept": "text/plain", - } - - try: - response = requests.get( - HEALTHCHECK_URL, headers=headers, verify=SSL_VERIFY, timeout=10 - ) - if response.status_code == 200: - print("Healthcheck successful") - return True - print(f"Healthcheck failed: HTTP {response.status_code}") - return False - except Exception as e: - print("Healthcheck error:", str(e)) - return False - - -def save_token_info(token_info): - with Path.open(TOKEN_FILE, "w") as f: - json.dump(token_info, f, indent=4) - - -def load_token_info(): - try: - with Path.open(TOKEN_FILE, "r") as f: - return json.load(f) - except (FileNotFoundError, json.JSONDecodeError): - return None - - -def main(): - token_info = load_token_info() - - # If no tokens exist or file is corrupted - if ( - not token_info - or "access_token" not in token_info - or "refresh_token" not in token_info - ): - print("No valid tokens found. Starting authentication...") - token_info = authenticate() - save_token_info(token_info) - healthcheck(token_info["access_token"]) - return - - # Check access token expiration with 10-second buffer - current_time = time.time() - if current_time < token_info["expires_at"] - 10: - print("Access token is valid") - healthcheck(token_info["access_token"]) - return - - print("Access token expired. Refreshing tokens...") - new_token_info = refresh_tokens(token_info["refresh_token"]) - - if new_token_info: - # Preserve existing refresh token if not updated - if "refresh_token" not in new_token_info: - new_token_info["refresh_token"] = token_info["refresh_token"] - - save_token_info(new_token_info) - healthcheck(new_token_info["access_token"]) - else: - print("Refresh token invalid. Starting re-authentication...") - token_info = authenticate() - save_token_info(token_info) - healthcheck(token_info["access_token"]) - - -if __name__ == "__main__": - main() From bea42cc65905304ab760fb43f89133fff6f7a8dc Mon Sep 17 00:00:00 2001 From: "i.alexandrov" Date: Tue, 9 Dec 2025 14:45:42 +0300 Subject: [PATCH 02/25] Change stand_cloud_thirdparty_integration path --- mkdocs.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/mkdocs.yml b/mkdocs.yml index ff4a8696..0d4d2c7b 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -81,7 +81,7 @@ nav: - Skip test: examples/skip_test.md - StandCloud: examples/stand_cloud.md - StandCloud read data: examples/stand_cloud_reader.md - - StandCloud third party authorization: examples/stand_cloud_thirdparty_auth.md + - StandCloud third party integration: examples/stand_cloud_thirdparty_integration.md - Stand equipment: examples/stand_equipment.md - About: - Development: about/development.md From 85ada86897c1cac0904eb0b9da9235dd3cfede30 Mon Sep 17 00:00:00 2001 From: "i.alexandrov" Date: Tue, 9 Dec 2025 14:45:56 +0300 Subject: [PATCH 03/25] [examples] Rename doc --- ...d_thirdparty_auth.md => stand_cloud_thirdparty_integration.md} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename docs/examples/{stand_cloud_thirdparty_auth.md => stand_cloud_thirdparty_integration.md} (100%) diff --git a/docs/examples/stand_cloud_thirdparty_auth.md b/docs/examples/stand_cloud_thirdparty_integration.md similarity index 100% rename from docs/examples/stand_cloud_thirdparty_auth.md rename to docs/examples/stand_cloud_thirdparty_integration.md From 6cd8b54337f13fc2f2f0996ed84515b8aca297db Mon Sep 17 00:00:00 2001 From: "i.alexandrov" Date: Tue, 9 Dec 2025 14:47:37 +0300 Subject: [PATCH 04/25] [examples] Rename integration_flow --- .../stand_cloud_integration/{auth_flow.py => integration_flow.py} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename examples/stand_cloud_integration/{auth_flow.py => integration_flow.py} (100%) diff --git a/examples/stand_cloud_integration/auth_flow.py b/examples/stand_cloud_integration/integration_flow.py similarity index 100% rename from examples/stand_cloud_integration/auth_flow.py rename to examples/stand_cloud_integration/integration_flow.py From f14770c9562669b5d88669a9e18d248839e6ab65 Mon Sep 17 00:00:00 2001 From: "i.alexandrov" Date: Tue, 9 Dec 2025 14:59:01 +0300 Subject: [PATCH 05/25] [stand_cloud_integration] Rm excess consts --- examples/stand_cloud_integration/integration_flow.py | 11 ++++------- 1 file changed, 4 insertions(+), 7 deletions(-) diff --git a/examples/stand_cloud_integration/integration_flow.py b/examples/stand_cloud_integration/integration_flow.py index 135ac2b5..9b5919ce 100644 --- a/examples/stand_cloud_integration/integration_flow.py +++ b/examples/stand_cloud_integration/integration_flow.py @@ -2,16 +2,14 @@ import requests +SSL_VERIFY = True + ######################################## # Change API_KEY to your api key API_KEY = "your_api_key" ######################################## -SSL_VERIFY = True -BASE_URL = "https://standcloud.io" -API_URL = BASE_URL + "/integration/api/v1" # API endpoints - -HEALTHCHECK_URL = API_URL + "/healthcheck" +HEALTHCHECK_URL = "https://standcloud.io/integration/api/v1/healthcheck" response = requests.get(HEALTHCHECK_URL, verify=SSL_VERIFY) if response.status_code != 200: @@ -25,9 +23,8 @@ "Content-type": "application/json", "Accept": "text/plain", } -print("\nheader: ", header) -USER_INFO_URL = API_URL + "/test_run" +USER_INFO_URL = "https://standcloud.io/integration/api/v1/test_run" response = requests.get(USER_INFO_URL, headers=header, verify=SSL_VERIFY) if response.status_code != 200: From b4cc2eb39c9ac47881cfae382a1067f55cb59f9f Mon Sep 17 00:00:00 2001 From: "i.alexandrov" Date: Tue, 9 Dec 2025 15:04:22 +0300 Subject: [PATCH 06/25] [examples] Update StandCloud third party integration article --- .../stand_cloud_thirdparty_integration.md | 525 +----------------- 1 file changed, 23 insertions(+), 502 deletions(-) diff --git a/docs/examples/stand_cloud_thirdparty_integration.md b/docs/examples/stand_cloud_thirdparty_integration.md index 94ee8e52..5c63df9c 100644 --- a/docs/examples/stand_cloud_thirdparty_integration.md +++ b/docs/examples/stand_cloud_thirdparty_integration.md @@ -1,535 +1,56 @@ -# StandCloud third party authorization +# StandCloud third party integration -???+ warning - The information on this page is currently out of date! - -This documentation describes how to authorize a **HardPy** application on a -**StandCloud** system using the OAuth 2.0 Device Authorization Flow process. - -## Authorization process - -1. The application requests the device code from the authorization server. -2. The server returns: - * `device_code` - * `user_code` - * `verification_uri` (authorization URL) - * `verification_uri_complete` (the URL with the pre-filled code) - * `expires_in` (code lifetime) - * `interval` (status polling interval) -3. The user follows the link and confirms authorization. -4. The application periodically polls the token server. -5. After confirmation, the server returns an access token and a refresh token. +This documentation explains how to use the StandCloud integration service via an API key. +Information on integration options with StandCloud can be found at +https://standcloud.io/integration/api/v1/docs. ## Code example The code samples are in Python, but they use simple constructs that are available in most programming languages. -To implement the authorization process for a third-party -application in a different language, use the general approach demonstrated in these scripts. -Reading [RFC6749](https://datatracker.ietf.org/doc/html/rfc8628), -which describes the OAuth 2.0 device flow, is also recommended. -The above examples do not display the QR code to the user; rather, they display a link for authorization. -## Authorization example +## Integration example -**auth_flow.py** is an example of a simple authorization script written in Python. +**integration_flow.py** is an example of a simple integration script written in Python. The script requires the installation of the `requests` package. The only thing that needs to be changed to make it work is the -**StandCloud** address in the `BASE_URL` variable. +**API key** in the `API_KEY` variable. +This script first checks access to the integration service, +and then requests a list of completed test runs by sending a request to `test_run`. -```python -# This script demonstrates the StandCloud authentication process. -# Change the URL to your StandCloud address, nothing else needs to be changed. -import json +```python import sys -import time import requests -######################################## -# Change URL to your StandCloud address -BASE_URL = "https://demo.standcloud.io" -######################################## - -# Confugirable parameters, False if the server is self-signed SSL_VERIFY = True -# OAuth client configuration -# client_id must be "hardpy-report-uploader" -CLIENT_ID = "hardpy-report-uploader" - -# API endpoints -API_URL = BASE_URL + "/hardpy/api/v1" -DEVICE_AUTHORIZATION_URL = BASE_URL + "/auth/api/oidc/device-authorization" -TOKEN_URL = BASE_URL + "/auth/api/oidc/token" - - -# Device authorization aequest -data = { - "client_id": CLIENT_ID, - "scope": "offline_access authelia.bearer.authz", - "audience": API_URL, -} +######################################## +# Change API_KEY to your api key +API_KEY = "your_api_key" +######################################## -req = requests.post(DEVICE_AUTHORIZATION_URL, data=data, verify=SSL_VERIFY, timeout=10) -response = json.loads(req.content) -verification_uri_complete = response["verification_uri_complete"] -interval = response["interval"] +HEALTHCHECK_URL = "https://standcloud.io/integration/api/v1/healthcheck" +response = requests.get(HEALTHCHECK_URL, verify=SSL_VERIFY) -# Token request -if "error" in response: - error = response["error"] - error_description = response["error_description"] - print(f"{error}: {error_description}") +if response.status_code != 200: + print(response.text) sys.exit(1) - -print(f"\nVerification URI: {verification_uri_complete}") - -data = { - "client_id": CLIENT_ID, - "grant_type": "urn:ietf:params:oauth:grant-type:device_code", - "device_code": response["device_code"], -} - -# Token response -while True: - req = requests.post(TOKEN_URL, data=data, verify=SSL_VERIFY, timeout=10) - response = json.loads(req.content) - print(".", end="") - if "access_token" in response: - print("\nToken info:", response) - break - time.sleep(interval) +print("StandCloud is up and running") # Test API call header = { - "Authorization": "Bearer {}".format(response["access_token"]), + "Authorization": f"Bearer {API_KEY}", "Content-type": "application/json", "Accept": "text/plain", } -print("\nheader: ", header) -USER_INFO_URL = API_URL + "/healthcheck" -response = requests.get(USER_INFO_URL, headers=header, verify=SSL_VERIFY, timeout=10) +USER_INFO_URL = "https://standcloud.io/integration/api/v1/test_run" +response = requests.get(USER_INFO_URL, headers=header, verify=SSL_VERIFY) if response.status_code != 200: print(response.text) sys.exit(1) -print("\nOAuth Authenticated and Authorized API access") +print("\nTest run list: ", response.text) ``` - -### 1. Introduction - -The script facilitates a secure way for devices with limited input capabilities -(like command-line applications) to authenticate with **StandCloud**. -It leverages the OAuth 2.0 Device Flow, which allows a user to authorize a device -by visiting a URL on a separate, input-rich device (e.g., a web browser on a computer or smartphone). - -### 2. Key concepts: OAuth 2.0 device authorization grant flow - -The OAuth 2.0 Device Authorization Grant Flow is designed for devices that cannot easily -display a browser or accept direct user input (like typing a username and password). -The flow generally involves the following steps: - -1. **Device Authorization Request**: The client (this script) requests a device code and a - verification URI from the authorization server. -2. **User Interaction**: The user takes the verification URI and a user code - (provided by the authorization server) and enters them into a browser on a - separate device to authorize the client. -3. **Token Request Polling**: The client repeatedly polls the authorization server's - token endpoint with the device code until the user completes the authorization. -4. **Token Response**: Once authorized, the client receives access tokens and (optionally) refresh tokens. -5. **API Access**: The client uses the obtained access token to make authenticated calls to the protected API resources. - -For a deeper understanding of the OAuth 2.0 Device Authorization Grant, refer to the official specification: -[RFC 8628 - OAuth 2.0 Device Authorization Grant](https://datatracker.ietf.org/doc/html/rfc8628). - -### 3. Script breakdown - -The script performs the following sequence of operations: - -#### 3.1. Configuration - -* **`BASE_URL`**: This variable must be changed to your specific **StandCloud** instance URL. - * **Example**: `https://company_name.standcloud.io` -* **`SSL_VERIFY`**: A boolean flag (`True` by default) indicating whether SSL certificate verification should be performed. - Set to `False` if your server uses a self-signed certificate. -* **`CLIENT_ID`**: The OAuth client identifier, which is fixed as `hardpy-report-uploader`. - This identifies the client application to the **StandCloud** authentication system. - -#### 3.2. API endpoints - -The script defines several critical API endpoints based on the `BASE_URL`: - -* **`API_URL`**: The base URL for the **StandCloud** API (`/hardpy/api/v1`). -* **`DEVICE_AUTHORIZATION_URL`**: The endpoint for initiating the device authorization flow (`/auth/api/oidc/device-authorization`). -* **`TOKEN_URL`**: The endpoint for requesting tokens after device authorization (`/auth/api/oidc/token`). - -#### 3.3. Device authorization request - -The script first makes a `POST` request to the `DEVICE_AUTHORIZATION_URL` with the following parameters: - -* **`client_id`**: The `CLIENT_ID` defined above. -* **`scope`**: Defines the permissions requested. - * `offline_access`: Allows the client to request refresh tokens, - enabling long-term access without re-authorization. - * `authelia.bearer.authz`: A specific scope related to Authelia - (an open-source authentication and authorization server often used with OpenID Connect) - for bearer token authorization. -* **`audience`**: Specifies the intended recipient of the access token, which is the `API_URL`. - -Upon a successful response, the script extracts: - -* **`verification_uri_complete`**: The full URL the user needs to visit in their browser to authorize the device. -* **`interval`**: The recommended polling interval (in seconds) for subsequent token requests. -* **`device_code`**: A code representing the authorization request, used in subsequent token polling. - -The script then prints the `verification_uri_complete` for the user to access. -It includes basic error handling for the device authorization request. - -#### 3.4. Token request (polling) - -After initiating the device authorization, the script enters a loop to poll -the `TOKEN_URL` until authorization is granted by the user. -Each `POST` request to the `TOKEN_URL` includes: - -* **`client_id`**: The `CLIENT_ID`. -* **`grant_type`**: Set to `"urn:ietf:params:oauth:grant-type:device_code"`, indicating the Device Flow grant type. -* **`device_code`**: The `device_code` obtained from the initial device authorization request. - -The script pauses for the `interval` period between each poll. -Once the `access_token` is present in the response, the polling loop breaks, and the script proceeds. - -#### 3.5. Test API call - -Finally, the script demonstrates how to use the obtained `access_token` to make an authenticated API call. - -* An `Authorization` header is constructed with the format `"Bearer "`. -* A `GET` request is made to the `API_URL` + `/healthcheck` endpoint. - This is a simple endpoint to verify successful authentication and authorization. - -If the API call returns a `200 OK` status, it confirms that the OAuth authentication and -authorization process was successful, and the client can now access protected resources. -Any other status code indicates an issue, and the response text is printed for debugging. - -### 4. Replicating in other languages - -To implement this authentication flow in a different programming language, you will need to: - -1. **HTTP Client Library**: Use an HTTP client library available in your chosen language - (e.g., `requests` for Python, `HttpClient` for C#, `fetch` for JavaScript, `OkHttp` for Java, `net/http` for Go). -2. **JSON Parsing**: Implement JSON parsing to handle the responses from the authorization and token endpoints. -3. **URL Construction**: Construct the `DEVICE_AUTHORIZATION_URL`, `TOKEN_URL`, and `API_URL` - correctly based on your **StandCloud** instance. -4. **Request Parameters**: Ensure that the `client_id`, `scope`, `audience`, `grant_type`, - and `device_code` parameters are correctly included in your `POST` request bodies, typically as form-urlencoded data. -5. **Polling Logic**: Implement a polling mechanism with appropriate delays to wait for user authorization. -6. **Bearer Token**: Correctly format the `Authorization: Bearer ` header for protected API calls. - -## Authorization example with token update - -**token_update.py** is a more complex version of the **auth_flow.py** script, -which rotates access and refresh tokens. -This script stores these tokens in a file. -However, this simplified token storage model should not be used in a production -environment for security reasons. -The only thing that needs to be changed to make it work is the -**StandCloud** address in the `BASE_URL` variable. - -```python -# This script demonstrates the StandCloud authentication process with -# token updating. Don't use the save as file option, it's insecure. -# Change the URL to your StandCloud address, nothing else needs to be changed. - -import json -import sys -import time -from pathlib import Path - -import requests - -######################################## -# Change URL to your StandCloud address -BASE_URL = "https://demo.standcloud.io" -######################################## - -# Configurable parameters, False if the server is self-signed -SSL_VERIFY = True - -# OAuth client configuration -# client_id must be "hardpy-report-uploader" -CLIENT_ID = "hardpy-report-uploader" - -# API endpoints -API_URL = BASE_URL + "/hardpy/api/v1" -DEVICE_AUTHORIZATION_URL = BASE_URL + "/auth/api/oidc/device-authorization" -TOKEN_URL = BASE_URL + "/auth/api/oidc/token" -HEALTHCHECK_URL = API_URL + "/healthcheck" -TOKEN_FILE = Path(__file__).parent / "token_info.json" - - -def authenticate(): - data = { - "client_id": CLIENT_ID, - "scope": "offline_access authelia.bearer.authz", - "audience": API_URL, - } - - req = requests.post( - DEVICE_AUTHORIZATION_URL, data=data, verify=SSL_VERIFY, timeout=10 - ) - response = req.json() - - if "error" in response: - error = response["error"] - error_description = response["error_description"] - print(f"{error}: {error_description}") - sys.exit(1) - - verification_uri_complete = response["verification_uri_complete"] - interval = response["interval"] - - print(f"\nVerification URI: {verification_uri_complete}") - - data = { - "client_id": CLIENT_ID, - "grant_type": "urn:ietf:params:oauth:grant-type:device_code", - "device_code": response["device_code"], - } - - while True: - req = requests.post(TOKEN_URL, data=data, verify=SSL_VERIFY, timeout=10) - response = req.json() - if "access_token" in response: - print("\nAuthentication successful") - break - time.sleep(interval) - - # Calculate expiration time - expires_at = time.time() + response["expires_in"] - token_info = { - "access_token": response["access_token"], - "refresh_token": response["refresh_token"], - "expires_at": expires_at, - } - return token_info - - -def refresh_tokens(refresh_token_value): - data = { - "client_id": CLIENT_ID, - "grant_type": "refresh_token", - "refresh_token": refresh_token_value, - } - - try: - req = requests.post(TOKEN_URL, data=data, verify=SSL_VERIFY, timeout=10) - response = req.json() - - if "access_token" not in response: - print( - "Token refresh failed. Error:", response.get("error", "unknown error") - ) - return None - - expires_at = time.time() + response["expires_in"] - token_info = { - "access_token": response["access_token"], - "expires_at": expires_at, - } - - # Update refresh token if new one is provided - if "refresh_token" in response: - token_info["refresh_token"] = response["refresh_token"] - print("Access and refresh tokens updated") - else: - print("Access token updated") - - return token_info - - except Exception as e: - print("Token refresh failed:", str(e)) - return None - - -def healthcheck(access_token): - headers = { - "Authorization": f"Bearer {access_token}", - "Content-type": "application/json", - "Accept": "text/plain", - } - - try: - response = requests.get( - HEALTHCHECK_URL, headers=headers, verify=SSL_VERIFY, timeout=10 - ) - if response.status_code == 200: - print("Healthcheck successful") - return True - print(f"Healthcheck failed: HTTP {response.status_code}") - return False - except Exception as e: - print("Healthcheck error:", str(e)) - return False - - -def save_token_info(token_info): - with Path.open(TOKEN_FILE, "w") as f: - json.dump(token_info, f, indent=4) - - -def load_token_info(): - try: - with Path.open(TOKEN_FILE, "r") as f: - return json.load(f) - except (FileNotFoundError, json.JSONDecodeError): - return None - - -def main(): - token_info = load_token_info() - - # If no tokens exist or file is corrupted - if ( - not token_info - or "access_token" not in token_info - or "refresh_token" not in token_info - ): - print("No valid tokens found. Starting authentication...") - token_info = authenticate() - save_token_info(token_info) - healthcheck(token_info["access_token"]) - return - - # Check access token expiration with 10-second buffer - current_time = time.time() - if current_time < token_info["expires_at"] - 10: - print("Access token is valid") - healthcheck(token_info["access_token"]) - return - - print("Access token expired. Refreshing tokens...") - new_token_info = refresh_tokens(token_info["refresh_token"]) - - if new_token_info: - # Preserve existing refresh token if not updated - if "refresh_token" not in new_token_info: - new_token_info["refresh_token"] = token_info["refresh_token"] - - save_token_info(new_token_info) - healthcheck(new_token_info["access_token"]) - else: - print("Refresh token invalid. Starting re-authentication...") - token_info = authenticate() - save_token_info(token_info) - healthcheck(token_info["access_token"]) - - -if __name__ == "__main__": - main() -``` - -### 1. Script breakdown - -This script extends the previous basic authentication by adding functions for token management. - -#### 1.1. Configuration and endpoints - -Most configurations are similar to the previous script: - -* **`BASE_URL`**: **StandCloud** instance URL. -* **`SSL_VERIFY`**: For SSL certificate validation. -* **`CLIENT_ID`**: Fixed as `"hardpy-report-uploader"`. -* **`API_URL`**: Base URL for API calls. -* **`DEVICE_AUTHORIZATION_URL`**: Endpoint for initiating device flow. -* **`TOKEN_URL`**: Endpoint for token requests (both initial and refresh). -* **`HEALTHCHECK_URL`**: A specific endpoint used to test API access. -* **`TOKEN_FILE`**: A `Path` object pointing to `token_info.json` in the same - directory as the script. This file is used to store token information for persistence. - **(Remember the security warning regarding file storage)**. - -#### 1.2. `authenticate()` function - -This function encapsulates the **initial OAuth 2.0 Device Authorization Grant Flow**, -identical to the previous script's main logic. - -1. It initiates a `POST` request to `DEVICE_AUTHORIZATION_URL` with `client_id`, - `scope` (`offline_access` is crucial here for getting a refresh token), and `audience`. -2. It handles the user interaction step, printing the `verification_uri_complete`. -3. It polls the `TOKEN_URL` using the `device_code` until an `access_token` - (and importantly, a `refresh_token`) is received. -4. It calculates the `expires_at` timestamp based on `expires_in` and the current time, - then returns a dictionary (`token_info`) containing `access_token`, `refresh_token`, and `expires_at`. - -#### 1.3. `refresh_tokens()` function - -This new function handles the **refresh token grant type**. - -1. It constructs a `POST` request to the `TOKEN_URL` with: - * **`client_id`**: Your client identifier. - * **`grant_type`**: Set to `"refresh_token"`. - * **`refresh_token`**: The `refresh_token_value` obtained from a previous authentication or refresh. -2. Upon a successful response, it extracts the `access_token` and its new `expires_at` timestamp. -3. Crucially, it checks if a *new* `refresh_token` is provided in the response. - If so, it updates the stored refresh token. Some authorization servers issue new - refresh tokens with each refresh; others do not. - Best practice is to use the newest one if provided. -4. It returns the `token_info` dictionary. - If the refresh fails, it prints an error and returns `None`. - -#### 1.4. `healthcheck()` function - -This helper function tests the validity of an `access_token` by making a `GET` request to the `HEALTHCHECK_URL`. - -1. It sets the `Authorization` header with the `Bearer` token. -2. It returns `True` if the response status is `200 OK`, - indicating successful API access, and `False` otherwise. - -#### 1.5. `save_token_info()` and `load_token_info()` functions - -These functions handle the **persistence of token information** to and from a local JSON file (`token_info.json`). - -* **`save_token_info`**: Writes the `token_info` dictionary to the `TOKEN_FILE`. -* **`load_token_info`**: Reads the `token_info` from the `TOKEN_FILE`. - It includes error handling for `FileNotFoundError` or `json.JSONDecodeError` - if the file doesn't exist or is corrupted, returning `None` in such cases. - -#### 1.6. `main()` - -This function orchestrates the script's logic. - -1. **Load Existing Tokens**: It first attempts to `load_token_info()` from the file. -2. **Initial Authentication**: If no valid tokens are found (file doesn't exist, - is corrupted, or essential keys are missing), it calls `authenticate()` - to start a new device flow, then `save_token_info()` and performs a `healthcheck`. -3. **Check Token Expiration**: If tokens are loaded, it checks if the `access_token` - is still valid (with a 10-second buffer before actual expiration). - * If valid, it performs a `healthcheck` directly. -4. **Token Refresh**: If the `access_token` has expired or is about to expire, - it calls `refresh_tokens()` using the stored `refresh_token`. - * If the refresh is successful, it updates the `token_info` - (preserving the refresh token if a new one wasn't issued), `save_token_info()`, and performs a `healthcheck`. - * If the refresh fails (e.g., refresh token is revoked or expired), - it falls back to starting a new full `authenticate()` flow. - -### 2. Replicating in other languages - -To implement this robust authentication and token refresh logic in another programming language, you'll need to: - -1. **HTTP Client**: Utilize your language's HTTP client for making `POST` and `GET` requests. -2. **JSON Handling**: Parse JSON responses for token information and API results. -3. **Time Management**: Accurately calculate token expiration times using timestamps. -4. **Token Storage**: Implement a **secure** method to store `access_token` and especially - `refresh_token` between application runs. Avoid plain text files in production. Consider: - * Environment variables (for secrets, typically during deployment). - * Operating system's credential manager (e.g., macOS Keychain, Windows Credential Manager). - * Encrypted database or file storage. - * Secure vaults (e.g., HashiCorp Vault, AWS Secrets Manager). -5. **Flow Logic**: - * **Initial Authentication**: Implement the Device Flow as described in the previous documentation. - * **Token Expiration Check**: Before making an API call, always check if the access token is near expiration. - * **Token Refresh Logic**: If expired, make a `POST` request to the `TOKEN_URL` with `grant_type=refresh_token` and the `refresh_token`. - * **Fallback to Re-authentication**: If the refresh token fails, initiate the full Device Flow again. -6. **Error Handling**: Implement robust error handling for network issues, - invalid responses, and authentication failures at each step. From 8f5c82f646070c5d1869d9888f75286989e2dcaa Mon Sep 17 00:00:00 2001 From: "i.alexandrov" Date: Wed, 10 Dec 2025 10:10:42 +0300 Subject: [PATCH 07/25] Add Discord invite link --- README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/README.md b/README.md index d65c1759..08a1c929 100644 --- a/README.md +++ b/README.md @@ -13,7 +13,7 @@ HardPy is a python library for creating a test bench for devices. [![pytest versions](https://img.shields.io/badge/pytest-%3E%3D7.0-blue)](https://docs.pytest.org/en/latest/) [![Documentation](https://img.shields.io/badge/Documentation%20-Overview%20-%20%23007ec6)](https://everypinio.github.io/hardpy/) [![Reddit](https://img.shields.io/badge/-Reddit-FF4500?style=flat&logo=reddit&logoColor=white)](https://www.reddit.com/r/HardPy) -[![Discord](https://img.shields.io/discord/1304494076799877172?color=7389D8&label&logo=discord&logoColor=ffffff)]() +[![Discord](https://img.shields.io/discord/1304494076799877172?color=7389D8&label&logo=discord&logoColor=ffffff)](https://discord.com/invite/3kBG9CbS) [![Telegram](https://img.shields.io/badge/-Telegram-2CA5E0?style=flat&logo=telegram&logoColor=white)](https://t.me/everypin) From a3dbe7c2af6193d5b3f73af990cf8fea1d22d6c1 Mon Sep 17 00:00:00 2001 From: "i.alexandrov" Date: Mon, 29 Dec 2025 10:34:38 +0300 Subject: [PATCH 08/25] Update package to 0.20.0 --- docs/changelog.md | 5 +++++ pyproject.toml | 2 +- 2 files changed, 6 insertions(+), 1 deletion(-) diff --git a/docs/changelog.md b/docs/changelog.md index 4af0ead4..158537b2 100644 --- a/docs/changelog.md +++ b/docs/changelog.md @@ -2,6 +2,11 @@ Versions follow [Semantic Versioning](https://semver.org/): `..`. +## 0.20.0 + +* Add JSON file storage support as alternative to CouchDB. + [[PR-239](https://github.com/everypinio/hardpy/pull/239)] + ## 0.19.1 * Add a `stop_time` validator function. If a case or module has a start time, diff --git a/pyproject.toml b/pyproject.toml index 1339636d..9875bef5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -3,7 +3,7 @@ [project] name = "hardpy" - version = "0.19.1" + version = "0.20.0" description = "HardPy library for device testing" license = "GPL-3.0-or-later" authors = [{ name = "Everypin", email = "info@everypin.io" }] From 5c8791bfb92300f819e844684365730b639c2d07 Mon Sep 17 00:00:00 2001 From: "i.alexandrov" Date: Mon, 29 Dec 2025 10:34:56 +0300 Subject: [PATCH 09/25] Add JSON info --- README.md | 4 ++-- docs/index.md | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index 08a1c929..85040570 100644 --- a/README.md +++ b/README.md @@ -13,7 +13,7 @@ HardPy is a python library for creating a test bench for devices. [![pytest versions](https://img.shields.io/badge/pytest-%3E%3D7.0-blue)](https://docs.pytest.org/en/latest/) [![Documentation](https://img.shields.io/badge/Documentation%20-Overview%20-%20%23007ec6)](https://everypinio.github.io/hardpy/) [![Reddit](https://img.shields.io/badge/-Reddit-FF4500?style=flat&logo=reddit&logoColor=white)](https://www.reddit.com/r/HardPy) -[![Discord](https://img.shields.io/discord/1304494076799877172?color=7389D8&label&logo=discord&logoColor=ffffff)](https://discord.com/invite/3kBG9CbS) +[![Discord](https://img.shields.io/discord/1304494076799877172?color=7389D8&label&logo=discord&logoColor=ffffff)](https://discord.gg/98bWadmG8J) [![Telegram](https://img.shields.io/badge/-Telegram-2CA5E0?style=flat&logo=telegram&logoColor=white)](https://t.me/everypin) @@ -26,7 +26,7 @@ HardPy allows you to: * Create test benches for devices using [pytest](https://docs.pytest.org/); * Use a browser to view, start, stop, and interact with tests; -* Store test results in the [CouchDB](https://couchdb.apache.org/) database; +* Store test results in the [CouchDB](https://couchdb.apache.org/) database or to simple JSON files; * Store test results on the [StandCloud](https://standcloud.io/) analytics platform.

diff --git a/docs/index.md b/docs/index.md index 8e084310..53e1c755 100644 --- a/docs/index.md +++ b/docs/index.md @@ -14,7 +14,7 @@ HardPy allows you to: * Create test benches for devices using [pytest](https://docs.pytest.org/); * Use a browser to view, start, stop, and interact with tests; -* Store test results in the [CouchDB](https://couchdb.apache.org/) database; +* Store test results in the [CouchDB](https://couchdb.apache.org/) database or to simple JSON files; * Store test results on the [StandCloud](https://standcloud.io/) analytics platform.

From bdd9ae600462157d2fecb034b033bf49cd0bca3e Mon Sep 17 00:00:00 2001 From: "i.alexandrov" Date: Mon, 29 Dec 2025 10:35:26 +0300 Subject: [PATCH 10/25] [documentation] Add storage type info --- docs/documentation/cli.md | 1 + docs/documentation/hardpy_config.md | 12 ++++++++++-- 2 files changed, 11 insertions(+), 2 deletions(-) diff --git a/docs/documentation/cli.md b/docs/documentation/cli.md index 8a7f21a4..4982a99f 100644 --- a/docs/documentation/cli.md +++ b/docs/documentation/cli.md @@ -53,6 +53,7 @@ More info in [hardpy config](./hardpy_config.md). | [default: no-sc-autosync] │ | [default: check-stand-cloud] │ │ --sc-api-key TEXT Specify a StandCloud API key. │ +│ --storage-type TEXT Specify a storage type. [default: couchdb] | │ --help Show this message and exit. │ ╰────────────────────────────────────────────────────────────────────────────────────────────────────────────╯ ``` diff --git a/docs/documentation/hardpy_config.md b/docs/documentation/hardpy_config.md index a509afec..08728cd5 100644 --- a/docs/documentation/hardpy_config.md +++ b/docs/documentation/hardpy_config.md @@ -30,7 +30,7 @@ tests_name = "My tests" current_test_config = "" [database] -storage_type = "json" +storage_type = "couchdb" user = "dev" password = "dev" host = "localhost" @@ -88,13 +88,21 @@ Storage type. The default is `couchdb`. - `couchdb`: Stores test results and measurements in a CouchDB database. Requires a running CouchDB instance. - `json`: Stores test results and measurements in local JSON files. No external database required. + Files are stored in the `.hardpy` directory in the root of the project. The user can change this value with the `hardpy init --storage-type` option. #### storage_path Path to the storage directory. The default is `.hardpy` in the root of the project. -The user can change this value with the `hardpy init --storage-path` option. Relative and absolute paths are supported. +The user can change this value in the hardpy.toml file using the `storage_path` option. +Both relative and absolute paths are supported. + +```toml +[database] +storage_type = "json" +storage_path = "result" +``` #### user From 64b231fdccba1f7d69d87cec5b17a12df1d79bb0 Mon Sep 17 00:00:00 2001 From: "i.alexandrov" Date: Mon, 29 Dec 2025 10:36:05 +0300 Subject: [PATCH 11/25] [hardpy] Add storage_type --- hardpy/cli/cli.py | 6 ++++++ hardpy/common/config.py | 7 +++++++ 2 files changed, 13 insertions(+) diff --git a/hardpy/cli/cli.py b/hardpy/cli/cli.py index 1fc1f5f0..425ca090 100644 --- a/hardpy/cli/cli.py +++ b/hardpy/cli/cli.py @@ -82,6 +82,10 @@ def init( # noqa: PLR0913 default=default_config.stand_cloud.api_key, help="Specify a StandCloud API key.", ), + storage_type: str = typer.Option( + default=default_config.database.storage_type.value, + help="Specify a storage type.", + ), ) -> None: """Initialize HardPy tests directory. @@ -100,6 +104,7 @@ def init( # noqa: PLR0913 sc_connection_only (bool): Flag to check StandCloud service availability sc_autosync (bool): Flag to enable StandCloud auto syncronization sc_api_key (str | None): StandCloud API key + storage_type (str): Storage type, "json" or "couchdb", "couchdb" by default """ dir_path = Path(Path.cwd() / tests_dir if tests_dir else "tests") config_manager = ConfigManager() @@ -116,6 +121,7 @@ def init( # noqa: PLR0913 sc_connection_only=sc_connection_only, sc_autosync=sc_autosync, sc_api_key=sc_api_key, + storage_type=storage_type, ) # create tests directory Path.mkdir(dir_path, exist_ok=True, parents=True) diff --git a/hardpy/common/config.py b/hardpy/common/config.py index cf997190..14b7ba7b 100644 --- a/hardpy/common/config.py +++ b/hardpy/common/config.py @@ -173,6 +173,7 @@ def init_config( # noqa: PLR0913 sc_connection_only: bool, sc_autosync: bool, sc_api_key: str, + storage_type: str, ) -> None: """Initialize the HardPy configuration. @@ -192,11 +193,17 @@ def init_config( # noqa: PLR0913 sc_autosync (bool): StandCloud auto syncronization. sc_api_key (str): StandCloud API key. """ + # try: + # _storage_type = + # except ValueError: + # msg = f"Invalid storage type {storage_type}" + # raise ValueError(msg) self._config.tests_name = tests_name self._config.frontend.host = frontend_host self._config.frontend.port = frontend_port self._config.frontend.language = frontend_language self._config.database.user = database_user + self._config.database.storage_type = StorageType(storage_type) self._config.database.password = database_password self._config.database.host = database_host self._config.database.port = database_port From f51036b0fce674f7a2d64bda05b28576b82317d4 Mon Sep 17 00:00:00 2001 From: "i.alexandrov" Date: Mon, 29 Dec 2025 10:36:18 +0300 Subject: [PATCH 12/25] [test_cli] Add storage_type test --- tests/test_cli/test_hardpy_init.py | 27 +++++++++++++++++++++++++++ 1 file changed, 27 insertions(+) diff --git a/tests/test_cli/test_hardpy_init.py b/tests/test_cli/test_hardpy_init.py index beee5073..0f23878d 100644 --- a/tests/test_cli/test_hardpy_init.py +++ b/tests/test_cli/test_hardpy_init.py @@ -11,11 +11,13 @@ frontend_no_default_host = "localhost1" frontend_no_default_port = "8001" stand_cloud_no_default_addr = "everypin1.standcloud.localhost" +storage_type_no_default = "json" db_default_port = "5984" frontend_default_host = "localhost" frontend_default_port = "8000" frontend_default_language = "en" +storage_type_default = "couchdb" def test_cli_init(tmp_path: Path): @@ -112,6 +114,31 @@ def test_cli_init_db_port(tmp_path: Path): ";port = 5985" in content ), "couchdb.ini does not contain the expected port." +def test_cli_init_storage_type_default(tmp_path: Path): + subprocess.run( + [*HARDPY_COMMAND, tmp_path], + check=True, + ) + hardpy_toml_path = tmp_path / "hardpy.toml" + with Path.open(hardpy_toml_path) as f: + content = f.read() + storage_type_info = f"""storage_type = "{storage_type_default}" +""" + assert_msg = "hardpy.toml does not contain the default storage type." + assert storage_type_info in content, assert_msg + +def test_cli_init_storage_type_no_default(tmp_path: Path): + subprocess.run( + [*HARDPY_COMMAND, tmp_path, "--storage-type", storage_type_no_default], + check=True, + ) + hardpy_toml_path = tmp_path / "hardpy.toml" + with Path.open(hardpy_toml_path) as f: + content = f.read() + storage_type_info = f"""storage_type = "{storage_type_no_default}" +""" + assert_msg = "hardpy.toml does not contain the valid storage type." + assert storage_type_info in content, assert_msg def test_cli_init_frontend_host(tmp_path: Path): subprocess.run( From 9b40b64d6daaf945c87108976069a5aa63ddcbc5 Mon Sep 17 00:00:00 2001 From: "i.alexandrov" Date: Mon, 29 Dec 2025 10:57:06 +0300 Subject: [PATCH 13/25] [examples] Add storage_type --- examples/attempts/hardpy.toml | 1 + examples/couchdb_load/hardpy.toml | 1 + examples/dialog_box/hardpy.toml | 1 + examples/hello_hardpy/hardpy.toml | 1 + examples/measurement/hardpy.toml | 1 + examples/minute_parity/hardpy.toml | 1 + examples/multiple_configs/hardpy.toml | 1 + examples/operator_msg/hardpy.toml | 1 + examples/stand_cloud/hardpy.toml | 1 + 9 files changed, 9 insertions(+) diff --git a/examples/attempts/hardpy.toml b/examples/attempts/hardpy.toml index b94f1a38..f6247fcf 100644 --- a/examples/attempts/hardpy.toml +++ b/examples/attempts/hardpy.toml @@ -3,6 +3,7 @@ tests_name = "Attempts" current_test_config = "" [database] +storage_type = "couchdb" user = "dev" password = "dev" host = "localhost" diff --git a/examples/couchdb_load/hardpy.toml b/examples/couchdb_load/hardpy.toml index a8c4bab9..a7001d31 100644 --- a/examples/couchdb_load/hardpy.toml +++ b/examples/couchdb_load/hardpy.toml @@ -3,6 +3,7 @@ tests_name = "CouchDB Load" current_test_config = "" [database] +storage_type = "couchdb" user = "dev" password = "dev" host = "localhost" diff --git a/examples/dialog_box/hardpy.toml b/examples/dialog_box/hardpy.toml index 6d502185..d305d1c0 100644 --- a/examples/dialog_box/hardpy.toml +++ b/examples/dialog_box/hardpy.toml @@ -3,6 +3,7 @@ tests_name = "Dialog Box" current_test_config = "" [database] +storage_type = "couchdb" user = "dev" password = "dev" host = "localhost" diff --git a/examples/hello_hardpy/hardpy.toml b/examples/hello_hardpy/hardpy.toml index ce541134..bb7559c6 100644 --- a/examples/hello_hardpy/hardpy.toml +++ b/examples/hello_hardpy/hardpy.toml @@ -3,6 +3,7 @@ tests_name = "Hello HardPy" current_test_config = "" [database] +storage_type = "couchdb" user = "dev" password = "dev" host = "localhost" diff --git a/examples/measurement/hardpy.toml b/examples/measurement/hardpy.toml index 022960ae..0c6b4902 100644 --- a/examples/measurement/hardpy.toml +++ b/examples/measurement/hardpy.toml @@ -3,6 +3,7 @@ tests_name = "Measurement" current_test_config = "" [database] +storage_type = "couchdb" user = "dev" password = "dev" host = "localhost" diff --git a/examples/minute_parity/hardpy.toml b/examples/minute_parity/hardpy.toml index 0342d5cd..f6c0fd9f 100644 --- a/examples/minute_parity/hardpy.toml +++ b/examples/minute_parity/hardpy.toml @@ -3,6 +3,7 @@ tests_name = "Minute Parity" current_test_config = "" [database] +storage_type = "couchdb" user = "dev" password = "dev" host = "localhost" diff --git a/examples/multiple_configs/hardpy.toml b/examples/multiple_configs/hardpy.toml index 89fd9a12..124a807e 100644 --- a/examples/multiple_configs/hardpy.toml +++ b/examples/multiple_configs/hardpy.toml @@ -9,6 +9,7 @@ host = "localhost" port = 5984 [frontend] +storage_type = "couchdb" host = "localhost" port = 8000 language = "en" diff --git a/examples/operator_msg/hardpy.toml b/examples/operator_msg/hardpy.toml index 89b4d452..d6dad4d0 100644 --- a/examples/operator_msg/hardpy.toml +++ b/examples/operator_msg/hardpy.toml @@ -3,6 +3,7 @@ tests_name = "Operator Message" current_test_config = "" [database] +storage_type = "couchdb" user = "dev" password = "dev" host = "localhost" diff --git a/examples/stand_cloud/hardpy.toml b/examples/stand_cloud/hardpy.toml index 3847a1aa..87f48d4a 100644 --- a/examples/stand_cloud/hardpy.toml +++ b/examples/stand_cloud/hardpy.toml @@ -3,6 +3,7 @@ tests_name = "StandCloud" current_test_config = "" [database] +storage_type = "couchdb" user = "dev" password = "dev" host = "localhost" From 95bf5b95f6b3339461cd05da905703452fbcd1fa Mon Sep 17 00:00:00 2001 From: "i.alexandrov" Date: Mon, 29 Dec 2025 11:00:25 +0300 Subject: [PATCH 14/25] [common] Add storage_type docstring --- hardpy/common/config.py | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/hardpy/common/config.py b/hardpy/common/config.py index 14b7ba7b..86f218f2 100644 --- a/hardpy/common/config.py +++ b/hardpy/common/config.py @@ -192,12 +192,8 @@ def init_config( # noqa: PLR0913 sc_connection_only (bool): StandCloud check availability. sc_autosync (bool): StandCloud auto syncronization. sc_api_key (str): StandCloud API key. + storage_type (str): Database storage type. """ - # try: - # _storage_type = - # except ValueError: - # msg = f"Invalid storage type {storage_type}" - # raise ValueError(msg) self._config.tests_name = tests_name self._config.frontend.host = frontend_host self._config.frontend.port = frontend_port From a4e527f042285f95aadf0c1abc42d43cdf907f21 Mon Sep 17 00:00:00 2001 From: "i.alexandrov" Date: Mon, 29 Dec 2025 11:44:33 +0300 Subject: [PATCH 15/25] [test_config] Add storage_type --- tests/test_common/test_config.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/tests/test_common/test_config.py b/tests/test_common/test_config.py index 440d07c5..54a3e2bb 100644 --- a/tests/test_common/test_config.py +++ b/tests/test_common/test_config.py @@ -64,10 +64,12 @@ def test_config_manager_init(): sc_connection_only=stand_cloud_no_default_connection_only, sc_autosync=stand_cloud_no_default_autosync, sc_api_key=stand_cloud_no_default_api_key, + storage_type="json", ) config = config_manager.config assert isinstance(config, HardpyConfig) assert config.tests_name == tests_no_default_name + assert config.database.storage_type == "json" assert config.database.user == db_no_default_user assert config.database.password == db_no_default_password assert config.database.host == db_no_default_host @@ -157,6 +159,7 @@ def test_config_manager_create_config(tmp_path: Path): sc_connection_only=stand_cloud_dafault_connection_only, sc_autosync=stand_cloud_default_autosync, sc_api_key=stand_cloud_default_api_key, + storage_type="json", ) config_manager.create_config(tests_dir) From cfa42ed4966a868444a51be2ccbf2433050bdac8 Mon Sep 17 00:00:00 2001 From: "i.alexandrov" Date: Mon, 29 Dec 2025 14:16:24 +0300 Subject: [PATCH 16/25] Add uuid6 for uuid7 using --- pyproject.toml | 1 + 1 file changed, 1 insertion(+) diff --git a/pyproject.toml b/pyproject.toml index 9875bef5..df36ea1b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -36,6 +36,7 @@ "tomli>=2.0.1, <3", "py-machineid~=0.6.0", "tzlocal~=5.2", + "uuid6", # Frontend "fastapi>=0.100.1", From d8636848bbb70c0d0a7d3d518418035d18f10066 Mon Sep 17 00:00:00 2001 From: "i.alexandrov" Date: Mon, 29 Dec 2025 14:17:04 +0300 Subject: [PATCH 17/25] [pytest_hardpy] Fix JSON tempstore usage --- hardpy/pytest_hardpy/db/tempstore.py | 12 ++++++++---- .../result/report_synchronizer/synchronizer.py | 11 ++++++++--- 2 files changed, 16 insertions(+), 7 deletions(-) diff --git a/hardpy/pytest_hardpy/db/tempstore.py b/hardpy/pytest_hardpy/db/tempstore.py index d729a3b5..235a08b7 100644 --- a/hardpy/pytest_hardpy/db/tempstore.py +++ b/hardpy/pytest_hardpy/db/tempstore.py @@ -8,7 +8,8 @@ from logging import getLogger from pathlib import Path from typing import TYPE_CHECKING -from uuid import uuid4 + +from uuid6 import uuid7 from hardpy.common.config import ConfigManager, StorageType from hardpy.common.singleton import SingletonMeta @@ -74,11 +75,12 @@ def __init__(self) -> None: config = ConfigManager() config_storage_path = Path(config.config.database.storage_path) if config_storage_path.is_absolute(): - self._storage_dir = config_storage_path / "tempstore" + self._storage_dir = config_storage_path / "storage" / "tempstore" else: self._storage_dir = Path( config.tests_path / config.config.database.storage_path + / "storage" / "tempstore", ) self._storage_dir.mkdir(parents=True, exist_ok=True) @@ -94,8 +96,10 @@ def push_report(self, report: ResultRunStore) -> bool: bool: True if successful, False otherwise """ report_dict = report.model_dump() - report_id = report_dict.get("id", str(uuid4())) - report_dict["id"] = report_id # Ensure ID is in the document + report_dict.pop("id", None) + report_id = str(uuid7()) + report_dict["_id"] = report_id + report_dict["_rev"] = report_id report_file = self._storage_dir / f"{report_id}.json" try: diff --git a/hardpy/pytest_hardpy/result/report_synchronizer/synchronizer.py b/hardpy/pytest_hardpy/result/report_synchronizer/synchronizer.py index 6d89b450..4d1c9b50 100644 --- a/hardpy/pytest_hardpy/result/report_synchronizer/synchronizer.py +++ b/hardpy/pytest_hardpy/result/report_synchronizer/synchronizer.py @@ -8,7 +8,7 @@ from pydantic import ValidationError from hardpy.common.stand_cloud.exception import StandCloudError -from hardpy.pytest_hardpy.db.tempstore import TempStore +from hardpy.pytest_hardpy.db.tempstore import CouchDBTempStore, TempStore from hardpy.pytest_hardpy.result.report_loader.stand_cloud_loader import ( StandCloudLoader, ) @@ -40,6 +40,7 @@ def sync(self) -> str: Returns: str: Synchronization message """ + _tempstore = self._get_tempstore if not self._get_tempstore.reports(): return "All reports are synchronized with StandCloud" loader = self._create_sc_loader() @@ -48,8 +49,12 @@ def sync(self) -> str: success_report_counter = 0 for _report in self._get_tempstore.reports(): try: - report_id: str = _report.get("id") # type: ignore[assignment] - document: dict = _report.get("doc") # type: ignore[assignment] + if isinstance(_tempstore, CouchDBTempStore): + document: dict = _report.get("doc") # type: ignore[assignment] + report_id: str = _report.get("id") # type: ignore[assignment] + else: + document: dict = _report + report_id: str = _report.get("_id") document.pop("rev") except KeyError: try: From d1293cb2a5f3c565f98356c28343489d2f645b4e Mon Sep 17 00:00:00 2001 From: "i.alexandrov" Date: Mon, 29 Dec 2025 15:34:48 +0300 Subject: [PATCH 18/25] [.vscode] Add JSON storage example --- .vscode/launch.json | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/.vscode/launch.json b/.vscode/launch.json index fc1522a4..e5b35449 100644 --- a/.vscode/launch.json +++ b/.vscode/launch.json @@ -138,6 +138,17 @@ "examples/minute_parity" ] }, + { + "name": "Python: Example JSON storage", + "type": "debugpy", + "request": "launch", + "module": "hardpy.cli.cli", + "console": "integratedTerminal", + "args": [ + "run", + "examples/json_storage" + ] + }, { "name": "Python: Example Multiple configs", "type": "debugpy", From 527cb28beb0f0d543f0cfb836662d91e5f9b0b63 Mon Sep 17 00:00:00 2001 From: "i.alexandrov" Date: Mon, 29 Dec 2025 15:34:59 +0300 Subject: [PATCH 19/25] Add json_storage.md --- mkdocs.yml | 1 + 1 file changed, 1 insertion(+) diff --git a/mkdocs.yml b/mkdocs.yml index 0d4d2c7b..cd720a89 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -70,6 +70,7 @@ nav: - Dialog box: examples/dialog_box.md - HardPy launch: examples/hardpy_launch.md - Hello HardPy: examples/hello_hardpy.md + - JSON storage: examples/json_storage.md - Launch arguments: examples/launch_arguments.md - Pytest logging: examples/logging.md - Measurement: examples/measurement.md From cb39cee170d82513456fda4d5267b181926ba24a Mon Sep 17 00:00:00 2001 From: "i.alexandrov" Date: Mon, 29 Dec 2025 15:35:22 +0300 Subject: [PATCH 20/25] Add example without a database --- README.md | 14 ++++++++++++++ docs/index.md | 14 ++++++++++++++ 2 files changed, 28 insertions(+) diff --git a/README.md b/README.md index 85040570..860d969b 100644 --- a/README.md +++ b/README.md @@ -41,6 +41,8 @@ pip install hardpy ## Getting Started +### With CouchDB + 1. Create your first test bench. ```bash hardpy init @@ -59,6 +61,18 @@ hardpy run Login and password: **dev**, database - **runstore**. +### Without a database + +1. Create your first test bench. +```bash +hardpy init --no-create-database --storage-type json +``` +2. Launch HardPy operator panel. +```bash +hardpy run +``` +3. View operator panel in browser: http://localhost:8000/ + ## Examples For more examples of using **HardPy**, see the [examples](https://github.com/everypinio/hardpy/tree/main/examples) folder and the [documentation](https://everypinio.github.io/hardpy/examples/). diff --git a/docs/index.md b/docs/index.md index 53e1c755..ebf9e2c4 100644 --- a/docs/index.md +++ b/docs/index.md @@ -29,6 +29,8 @@ pip install hardpy ## Getting Started +### With CouchDB + 1. Create your first test bench. ```bash @@ -62,6 +64,18 @@ hardpy run alt="hardpy runstore" style="width:500px;">

+### Without a database + +1. Create your first test bench. +```bash +hardpy init --no-create-database --storage-type json +``` +2. Launch HardPy operator panel. +```bash +hardpy run +``` +3. View operator panel in browser: http://localhost:8000/ + ## Measurement instruments **HardPy** does not contain any drivers for interacting with measuring equipment. From 656316449a2fee1675a90fa220090a9406152efb Mon Sep 17 00:00:00 2001 From: "i.alexandrov" Date: Mon, 29 Dec 2025 15:35:38 +0300 Subject: [PATCH 21/25] [docs] Fix link --- docs/changelog.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/changelog.md b/docs/changelog.md index 158537b2..1d0c753d 100644 --- a/docs/changelog.md +++ b/docs/changelog.md @@ -24,7 +24,7 @@ Versions follow [Semantic Versioning](https://semver.org/): `.. Date: Mon, 29 Dec 2025 15:36:05 +0300 Subject: [PATCH 22/25] [docs] Add JSON features --- docs/documentation/pytest_hardpy.md | 29 ++ docs/examples/json_storage.md | 408 ++++++++++++++++++++++++++++ docs/features/features.md | 11 +- 3 files changed, 447 insertions(+), 1 deletion(-) create mode 100644 docs/examples/json_storage.md diff --git a/docs/documentation/pytest_hardpy.md b/docs/documentation/pytest_hardpy.md index 984d66f3..fe6e2bde 100644 --- a/docs/documentation/pytest_hardpy.md +++ b/docs/documentation/pytest_hardpy.md @@ -911,6 +911,35 @@ def fill_actions_after_test(post_run_functions: list): yield ``` +#### JsonLoader + +Used to write reports to the JSON. + +**Arguments:** + +- `storage_dir` *(Path | None)*: JSON file directory. + +**Functions:** + +- `load` *(ResultRunStore, new_report_id)*: Load report to the JSON file. + +**Example:** + +```python +# conftest +def save_report_to_dir(): + report = get_current_report() + if report: + loader = JsonLoader(Path.cwd() / "reports") + loader.load(report) + + +@pytest.fixture(scope="session", autouse=True) +def fill_actions_after_test(post_run_functions: list): + post_run_functions.append(save_report_to_dir) + yield +``` + #### StandCloudLoader Used to write reports to the **StandCloud**. diff --git a/docs/examples/json_storage.md b/docs/examples/json_storage.md new file mode 100644 index 00000000..8a783f92 --- /dev/null +++ b/docs/examples/json_storage.md @@ -0,0 +1,408 @@ +# JSON storage + +This is an example of using **pytest-hardpy** functions, storing +the result to JSON file. +The code for this example can be seen inside the hardpy package +[JSON storage](https://github.com/everypinio/hardpy/tree/main/examples/json_storage). + +### how to start + +1. Launch `hardpy init --no-create-database --storage-type json json_storage`. +2. Modify the files described below. +3. Launch `hardpy run json_storage`. + +### hardpy.toml + +Replace the settings in the `[frontend]` and `[frontend.modal_result]` sections +with those shown in the **hardpy.toml** example file below. + +```toml +title = "HardPy JSON Storage Demo" +tests_name = "Device Test Suite" + +[database] +storage_type = "json" +storage_path = "result" + +[frontend] +host = "localhost" +port = 8000 +language = "en" +sound_on = false +full_size_button = false +manual_collect = false +measurement_display = true + +[frontend.modal_result] +enable = true +auto_dismiss_pass = true +auto_dismiss_timeout = 5 +``` + +### conftest.py + +```python +from pathlib import Path +import pytest + +from hardpy import JsonLoader, get_current_report + + +@pytest.fixture(scope="session") +def setup_test_environment(): + """Set up test environment before all tests.""" + print("\n=== Setting up test environment ===") + # Add any global setup here + yield + print("\n=== Tearing down test environment ===") + # Add any global cleanup here + + +@pytest.fixture(scope="function") +def test_device(): + """Fixture providing simulated test device.""" + + class TestDevice: + def __init__(self): + self.connected = False + self.voltage = 5.0 + self.current = 0.0 + + def connect(self): + self.connected = True + return True + + def disconnect(self): + self.connected = False + + def measure_voltage(self): + return self.voltage + + def measure_current(self): + return self.current + + device = TestDevice() + device.connect() + + yield device + + device.disconnect() + + +def save_report_to_dir(): + report = get_current_report() + if report: + loader = JsonLoader(Path.cwd() / "reports") + loader.load(report) + + +@pytest.fixture(scope="session", autouse=True) +def fill_actions_after_test(post_run_functions: list): + post_run_functions.append(save_report_to_dir) + yield +``` + +### test_chart_demo.py + +```python +import hardpy +import pytest +import math + +@pytest.mark.case_name("Sine Wave Analysis") +@pytest.mark.module_name("Chart Demonstrations") +def test_sine_wave(): + """Test generating and analyzing a sine wave.""" + hardpy.set_message("Generating sine wave data...") + + # Generate sine wave data + x_data = [] + y_data = [] + + for i in range(100): + x = i / 10.0 # 0 to 10 + y = math.sin(x) + x_data.append(x) + y_data.append(y) + + # Create chart + chart = hardpy.Chart( + title="Sine Wave", + x_label="Time", + y_label="Amplitude", + type=hardpy.ChartType.LINE, + ) + chart.add_series(x_data, y_data, "Sine Wave") + + hardpy.set_case_chart(chart) + + # Verify amplitude + max_amplitude = max(y_data) + min_amplitude = min(y_data) + peak_to_peak = max_amplitude - min_amplitude + + hardpy.set_case_measurement( + hardpy.NumericMeasurement( + name="Peak-to-Peak Amplitude", + value=peak_to_peak, + unit="V", + lower_limit=1.9, + upper_limit=2.1, + ) + ) + + hardpy.set_message(f"Peak-to-peak amplitude: {peak_to_peak:.3f}V") + + assert 1.9 <= peak_to_peak <= 2.1, "Amplitude out of range" + +@pytest.mark.case_name("Temperature Curve") +@pytest.mark.module_name("Chart Demonstrations") +def test_temperature_curve(): + """Test temperature rise curve.""" + hardpy.set_message("Recording temperature curve...") + + # Simulate temperature rise + time_data = [] + temp_data = [] + + for i in range(50): + time = i * 2 # seconds + # Exponential rise to 80°C + temp = 25 + 55 * (1 - math.exp(-i / 20)) + time_data.append(time) + temp_data.append(temp) + + # Create chart + chart = hardpy.Chart( + title="Temperature Rise Curve", + x_label="Time (seconds)", + y_label="Temperature (°C)", + type=hardpy.ChartType.LINE, + ) + chart.add_series(time_data, temp_data, "Temperature") + + hardpy.set_case_chart(chart) + + # Check final temperature + final_temp = temp_data[-1] + + hardpy.set_case_measurement( + hardpy.NumericMeasurement( + name="Final Temperature", + value=final_temp, + unit="°C", + upper_limit=85, + ) + ) + + hardpy.set_message(f"Final temperature: {final_temp:.1f}°C") + + assert final_temp < 85, f"Temperature too high: {final_temp}°C" +``` + +### test_communication.py + +```python +import hardpy +import pytest +from time import sleep + +@pytest.mark.case_name("Serial Port Connection") +@pytest.mark.module_name("Communication Tests") +def test_serial_connection(): + """Test serial port connection.""" + hardpy.set_message("Testing serial port connection...") + + # Simulate connection + port = "/dev/ttyUSB0" + baudrate = 115200 + + hardpy.set_instrument( + hardpy.Instrument( + name="Serial Port", + comment=f"{port} @ {baudrate} baud" + ) + ) + + # Simulate successful connection + connection_ok = True + + hardpy.set_message(f"Connected to {port} at {baudrate} baud") + + assert connection_ok, "Failed to establish serial connection" + +@pytest.mark.case_name("Data Transfer Test") +@pytest.mark.module_name("Communication Tests") +@pytest.mark.attempt(3) # Allow 2 retries +def test_data_transfer(): + """Test data transfer over serial.""" + hardpy.set_message("Testing data transfer...") + + # Simulate sending and receiving data + sent_bytes = 1024 + received_bytes = 1024 + transfer_time = 0.5 # seconds + + # Calculate transfer rate + transfer_rate = (sent_bytes + received_bytes) / transfer_time + + hardpy.set_case_measurement( + hardpy.NumericMeasurement( + name="Transfer Rate", + value=transfer_rate, + unit="bytes/s", + lower_limit=1000, + ) + ) + + hardpy.set_message(f"Transfer rate: {transfer_rate:.0f} bytes/s") + + assert received_bytes == sent_bytes, "Data integrity error" + assert transfer_rate > 1000, "Transfer rate too slow" + +@pytest.mark.case_name("Protocol Validation") +@pytest.mark.module_name("Communication Tests") +@pytest.mark.critical # Critical test - stops all if fails +def test_protocol_validation(): + """Test communication protocol validation.""" + hardpy.set_message("Validating communication protocol...") + + # Simulate protocol check + protocol_version = "v2.1" + expected_version = "v2.1" + + hardpy.set_case_measurement( + hardpy.StringMeasurement( + name="Protocol Version", + value=protocol_version, + comparison_value=expected_version, + ) + ) + + hardpy.set_message(f"Protocol version: {protocol_version}") + + assert protocol_version == expected_version, \ + f"Protocol mismatch: got {protocol_version}, expected {expected_version}" + +@pytest.mark.case_name("Error Handling Test") +@pytest.mark.module_name("Communication Tests") +@pytest.mark.dependency("test_communication::test_protocol_validation") +def test_error_handling(): + """Test error handling (depends on protocol validation).""" + hardpy.set_message("Testing error handling...") + + # Simulate error injection and recovery + errors_injected = 5 + errors_handled = 5 + + hardpy.set_case_measurement( + hardpy.NumericMeasurement( + name="Errors Handled", + value=errors_handled, + unit="count", + comparison_value=errors_injected, + ) + ) + + hardpy.set_message(f"Handled {errors_handled}/{errors_injected} errors") + + assert errors_handled == errors_injected, "Some errors not handled correctly" +``` + +### test_voltage.py + +```python +import hardpy +import pytest +from time import sleep + +@pytest.mark.case_name("Check Power Supply Voltage") +@pytest.mark.module_name("Power Supply Tests") +def test_power_supply_voltage(): + """Test that power supply outputs correct voltage.""" + # Set test stand information + hardpy.set_stand_name("Test Bench #1") + hardpy.set_stand_location("Lab A") + + # Set device under test information + hardpy.set_dut_serial_number("PSU-12345") + hardpy.set_dut_name("Power Supply Unit") + hardpy.set_dut_type("DC Power Supply") + + # Simulate voltage measurement + expected_voltage = 5.0 + measured_voltage = 5.02 # Simulated measurement + tolerance = 0.1 + + # Record measurement + hardpy.set_case_measurement( + hardpy.NumericMeasurement( + name="Output Voltage", + value=measured_voltage, + unit="V", + lower_limit=expected_voltage - tolerance, + upper_limit=expected_voltage + tolerance, + ) + ) + + # Add message + hardpy.set_message(f"Measured voltage: {measured_voltage}V (expected: {expected_voltage}V)") + + # Verify voltage is within tolerance + assert abs(measured_voltage - expected_voltage) <= tolerance, \ + f"Voltage out of tolerance: {measured_voltage}V" + +@pytest.mark.case_name("Check Current Limit") +@pytest.mark.module_name("Power Supply Tests") +def test_current_limit(): + """Test that power supply has correct current limit.""" + # Simulate current limit test + expected_limit = 3.0 + measured_limit = 3.05 # Simulated measurement + tolerance = 0.2 + + # Record measurement + hardpy.set_case_measurement( + hardpy.NumericMeasurement( + name="Current Limit", + value=measured_limit, + unit="A", + lower_limit=expected_limit - tolerance, + upper_limit=expected_limit + tolerance, + ) + ) + + hardpy.set_message(f"Current limit: {measured_limit}A") + + assert abs(measured_limit - expected_limit) <= tolerance + +@pytest.mark.case_name("Voltage Stability Test") +@pytest.mark.module_name("Power Supply Tests") +@pytest.mark.attempt(2) # Retry once if fails +def test_voltage_stability(): + """Test voltage stability over time.""" + hardpy.set_message("Testing voltage stability over 5 seconds...") + + voltage_readings = [] + for i in range(5): + # Simulate reading voltage + voltage = 5.0 + (i * 0.01) # Slight increase + voltage_readings.append(voltage) + sleep(0.1) # Simulate measurement delay + + max_variation = max(voltage_readings) - min(voltage_readings) + + # Record measurement + hardpy.set_case_measurement( + hardpy.NumericMeasurement( + name="Voltage Variation", + value=max_variation, + unit="V", + upper_limit=0.1, + ) + ) + + hardpy.set_message(f"Max voltage variation: {max_variation:.3f}V") + + assert max_variation < 0.1, f"Voltage not stable: {max_variation}V variation" +``` diff --git a/docs/features/features.md b/docs/features/features.md index d3f9f936..9f17e76f 100644 --- a/docs/features/features.md +++ b/docs/features/features.md @@ -190,7 +190,7 @@ The user can check the status of tests using the [hardpy status](./../documentat ### Storing test result in database -**HardPy** does not allow you to run tests without a running [CouchDB](https://couchdb.apache.org/) database. +**HardPy** allows you to run tests with a running [CouchDB](https://couchdb.apache.org/) database. This is a NoSQL database that ensures that the results of the current test run are committed, even if the tests are aborted early. @@ -206,6 +206,15 @@ An example of configuring **conftest.py** to store test run history can be found including the [couchdb_load](./../examples/couchdb_load.md) and [minute_parity](./../examples/minute_parity.md). +### JSON + +With **HardPy**, you can run tests without a database and save the test data to local JSON documents. +These documents have a structure similar to that of databases and documents in **CouchDB**. +A .hardpy/storage folder is created by default in the project folder, +where test reports can be found in the **runstore** folder. + +An example of its use can be found on page [JSON storage](./../examples/json_storage.md). + ### Other databases In order to save data to other databases, users will need to write their own adapter class to convert their From 49582e9e0d4836f5d0ec7aafe46c0e5065e1d3ab Mon Sep 17 00:00:00 2001 From: "i.alexandrov" Date: Mon, 29 Dec 2025 15:37:09 +0300 Subject: [PATCH 23/25] [json_storage] Mv json storage example --- examples/json_storage/.gitignore | 1 + examples/json_storage/README.md | 3 +++ .../conftest.py | 17 +++++++++++++++++ .../hardpy.toml | 0 .../pytest.ini | 0 .../test_chart_demo.py | 0 .../test_communication.py | 0 .../test_voltage.py | 0 8 files changed, 21 insertions(+) create mode 100644 examples/json_storage/.gitignore create mode 100644 examples/json_storage/README.md rename examples/{demo_json_storage => json_storage}/conftest.py (70%) rename examples/{demo_json_storage => json_storage}/hardpy.toml (100%) rename examples/{demo_json_storage => json_storage}/pytest.ini (100%) rename examples/{demo_json_storage => json_storage}/test_chart_demo.py (100%) rename examples/{demo_json_storage => json_storage}/test_communication.py (100%) rename examples/{demo_json_storage => json_storage}/test_voltage.py (100%) diff --git a/examples/json_storage/.gitignore b/examples/json_storage/.gitignore new file mode 100644 index 00000000..c4a847d9 --- /dev/null +++ b/examples/json_storage/.gitignore @@ -0,0 +1 @@ +/result diff --git a/examples/json_storage/README.md b/examples/json_storage/README.md new file mode 100644 index 00000000..dcc71eaf --- /dev/null +++ b/examples/json_storage/README.md @@ -0,0 +1,3 @@ +# JSON storage + +Example documentation: https://everypinio.github.io/hardpy/examples/json_storage/ diff --git a/examples/demo_json_storage/conftest.py b/examples/json_storage/conftest.py similarity index 70% rename from examples/demo_json_storage/conftest.py rename to examples/json_storage/conftest.py index 4c60935a..dd169b15 100644 --- a/examples/demo_json_storage/conftest.py +++ b/examples/json_storage/conftest.py @@ -1,5 +1,8 @@ +from pathlib import Path import pytest +from hardpy import JsonLoader, get_current_report + @pytest.fixture(scope="session") def setup_test_environment(): @@ -14,6 +17,7 @@ def setup_test_environment(): @pytest.fixture(scope="function") def test_device(): """Fixture providing simulated test device.""" + class TestDevice: def __init__(self): self.connected = False @@ -39,3 +43,16 @@ def measure_current(self): yield device device.disconnect() + + +def save_report_to_dir(): + report = get_current_report() + if report: + loader = JsonLoader(Path.cwd() / "reports") + loader.load(report) + + +@pytest.fixture(scope="session", autouse=True) +def fill_actions_after_test(post_run_functions: list): + post_run_functions.append(save_report_to_dir) + yield diff --git a/examples/demo_json_storage/hardpy.toml b/examples/json_storage/hardpy.toml similarity index 100% rename from examples/demo_json_storage/hardpy.toml rename to examples/json_storage/hardpy.toml diff --git a/examples/demo_json_storage/pytest.ini b/examples/json_storage/pytest.ini similarity index 100% rename from examples/demo_json_storage/pytest.ini rename to examples/json_storage/pytest.ini diff --git a/examples/demo_json_storage/test_chart_demo.py b/examples/json_storage/test_chart_demo.py similarity index 100% rename from examples/demo_json_storage/test_chart_demo.py rename to examples/json_storage/test_chart_demo.py diff --git a/examples/demo_json_storage/test_communication.py b/examples/json_storage/test_communication.py similarity index 100% rename from examples/demo_json_storage/test_communication.py rename to examples/json_storage/test_communication.py diff --git a/examples/demo_json_storage/test_voltage.py b/examples/json_storage/test_voltage.py similarity index 100% rename from examples/demo_json_storage/test_voltage.py rename to examples/json_storage/test_voltage.py From 79a6de6d520ea9a6fb01a88d9793e67534189121 Mon Sep 17 00:00:00 2001 From: "i.alexandrov" Date: Mon, 29 Dec 2025 15:37:40 +0300 Subject: [PATCH 24/25] [hardpy] Add JsonLoader --- hardpy/__init__.py | 2 + hardpy/pytest_hardpy/result/__init__.py | 2 + .../result/report_loader/json_loader.py | 49 +++++++++++++++++++ 3 files changed, 53 insertions(+) create mode 100644 hardpy/pytest_hardpy/result/report_loader/json_loader.py diff --git a/hardpy/__init__.py b/hardpy/__init__.py index ca4560b6..f11f2dda 100644 --- a/hardpy/__init__.py +++ b/hardpy/__init__.py @@ -45,6 +45,7 @@ ) from hardpy.pytest_hardpy.result import ( CouchdbLoader, + JsonLoader, StandCloudLoader, StandCloudReader, ) @@ -82,6 +83,7 @@ "HTMLComponent", "ImageComponent", "Instrument", + "JsonLoader", "MultistepWidget", "NumericInputWidget", "NumericMeasurement", diff --git a/hardpy/pytest_hardpy/result/__init__.py b/hardpy/pytest_hardpy/result/__init__.py index a82d101b..204629db 100644 --- a/hardpy/pytest_hardpy/result/__init__.py +++ b/hardpy/pytest_hardpy/result/__init__.py @@ -2,6 +2,7 @@ # GNU General Public License v3.0 (see LICENSE or https://www.gnu.org/licenses/gpl-3.0.txt) from hardpy.pytest_hardpy.result.report_loader.couchdb_loader import CouchdbLoader +from hardpy.pytest_hardpy.result.report_loader.json_loader import JsonLoader from hardpy.pytest_hardpy.result.report_loader.stand_cloud_loader import ( StandCloudLoader, ) @@ -13,6 +14,7 @@ __all__ = [ "CouchdbLoader", "CouchdbReader", + "JsonLoader", "StandCloudLoader", "StandCloudReader", ] diff --git a/hardpy/pytest_hardpy/result/report_loader/json_loader.py b/hardpy/pytest_hardpy/result/report_loader/json_loader.py new file mode 100644 index 00000000..c6e3d2eb --- /dev/null +++ b/hardpy/pytest_hardpy/result/report_loader/json_loader.py @@ -0,0 +1,49 @@ +# Copyright (c) 2024 Everypin +# GNU General Public License v3.0 (see LICENSE or https://www.gnu.org/licenses/gpl-3.0.txt) +from __future__ import annotations + +import json +from logging import getLogger +from pathlib import Path +from typing import TYPE_CHECKING + +from uuid6 import uuid7 + +if TYPE_CHECKING: + from hardpy.pytest_hardpy.db.schema import ResultRunStore + + +class JsonLoader: + """JSON report generator.""" + + def __init__(self, storage_dir: Path | None = None) -> None: + if not storage_dir: + storage_dir = Path.cwd() / "reports" + self._storage_dir = storage_dir + self._storage_dir.mkdir(parents=True, exist_ok=True) + self._log = getLogger(__name__) + + def load(self, report: ResultRunStore, new_report_id: str | None = None) -> bool: + """Load report to the report database. + + Args: + report (ResultRunStore): report + new_report_id (str | None, optional): user's report ID. Defaults to uuid7. + + Returns: + bool: True if success, else False + """ + report_dict = report.model_dump() + report_id = new_report_id if new_report_id else str(uuid7()) + report_dict["id"] = report_id + report_file = self._storage_dir / f"{report_id}.json" + + try: + with report_file.open("w") as f: + json.dump(report_dict, f, indent=2, default=str) + except Exception as exc: # noqa: BLE001 + self._log.error(f"Error while saving report {report_id}: {exc}") + return False + else: + self._log.debug(f"Report saved with id: {report_id}") + return True From 9f6a2a5f0697521d4c48bf0bee628896972c7548 Mon Sep 17 00:00:00 2001 From: "i.alexandrov" Date: Mon, 29 Dec 2025 15:59:37 +0300 Subject: [PATCH 25/25] [test_plugin] Add JSON tests --- tests/test_plugin/conftest.py | 36 +++++++++++++++---------- tests/test_plugin/hardpy.toml | 1 + tests/test_plugin/json_toml/hardpy.toml | 18 +++++++++++++ 3 files changed, 41 insertions(+), 14 deletions(-) create mode 100644 tests/test_plugin/json_toml/hardpy.toml diff --git a/tests/test_plugin/conftest.py b/tests/test_plugin/conftest.py index b386beb8..1d9b5f7f 100644 --- a/tests/test_plugin/conftest.py +++ b/tests/test_plugin/conftest.py @@ -7,18 +7,26 @@ pytest_plugins = "pytester" -@pytest.fixture -def hardpy_opts(): +@pytest.fixture(params=["couchdb", "json"], autouse=True) +def hardpy_opts(request): # noqa: ANN001 config_manager = ConfigManager() - config_data = config_manager.read_config( - Path(__file__).parent.resolve(), - ) - if not config_data: - msg = "Config not found" - raise RuntimeError(msg) - return [ - "--hardpy-clear-database", - "--hardpy-db-url", - config_data.database.url, - "--hardpy-pt", - ] + if request.param == "couchdb": + config_data = config_manager.read_config( + Path(__file__).parent.resolve(), + ) + if not config_data: + msg = "Config not found" + raise RuntimeError(msg) + + return [ + "--hardpy-clear-database", + "--hardpy-db-url", + config_data.database.url, + "--hardpy-pt", + ] + if request.param == "json": + config_data = config_manager.read_config( + Path(__file__).parent / "json_toml", + ) + return [ "--hardpy-clear-database", "--hardpy-pt"] + return None diff --git a/tests/test_plugin/hardpy.toml b/tests/test_plugin/hardpy.toml index c91e7d77..8f0c2ad0 100644 --- a/tests/test_plugin/hardpy.toml +++ b/tests/test_plugin/hardpy.toml @@ -1,6 +1,7 @@ title = "HardPy TOML config" [database] +storage_type = "couchdb" user = "dev" password = "dev" host = "localhost" diff --git a/tests/test_plugin/json_toml/hardpy.toml b/tests/test_plugin/json_toml/hardpy.toml new file mode 100644 index 00000000..86d5114b --- /dev/null +++ b/tests/test_plugin/json_toml/hardpy.toml @@ -0,0 +1,18 @@ +title = "HardPy TOML config" + +[database] +storage_type = "json" + +[frontend] +host = "localhost" +port = 8000 +language = "en" +full_size_button = false +sound_on = false +measurement_display = true +manual_collect = false + +[frontend.modal_result] +enable = false +auto_dismiss_pass = true +auto_dismiss_timeout = 5