diff --git a/app/methods/person_methods.py b/app/methods/person_methods.py index d6085b8..5c25d6c 100644 --- a/app/methods/person_methods.py +++ b/app/methods/person_methods.py @@ -1856,21 +1856,39 @@ def handle_email_person_auth_key_url( # subject = f'{account_short_name}: One Time Use Create Account Link ({new_auth_key})' body_html = f""" -

{to_name},

- -

If you did not request this account creation link, please delete this email. It is suggested that you delete this email after the account creation link has been used or if a new link has been requested.

- -

The link below can only be used once. If you would like try again using this method, you must request a new account creation link. If you request multiple links, only the newest link will work.

- -

Click to Finish Account Creation With One Time Use Link

- -

Or copy and paste the link:
- {person_auth_key_url}

- -

If you have questions about this email or trouble with this one time use link, you can email {help_tech_name} ({help_tech_email}).

- -

Thank you!

- """ +
+
+

{account_short_name}

+
+
+ +

Hello {to_name},

+ +

You have requested a one-time use link to complete your account registration. This link will allow you to set up your account securely.

+ +
+ + Finish Account Creation + +
+ +

+ Security Note: If you did not request this link, please delete this email. The link above can only be used once. If you request multiple links, only the newest one will be active. +

+ +
+

If the button above doesn't work, copy and paste the following URL into your browser:

+

+ {person_auth_key_url} +

+ +

+ Questions or trouble? Contact {help_tech_name}. +

+

Thank you!

+
+
+ """ if send_email(from_email=from_email, from_name=from_name, to_email=to_email, to_name=to_name, bcc_email=bcc_email, bcc_name=bcc_name, subject=subject, body_text=None, body_html=body_html): log.info(f'An email with a one time use sign in link was sent to {to_email}.') diff --git a/tests/check_db_schema.py b/tests/check_db_schema.py new file mode 100644 index 0000000..ee4c29a --- /dev/null +++ b/tests/check_db_schema.py @@ -0,0 +1,13 @@ +from app.db_sql import db +from sqlalchemy import text + +def check_columns(): + try: + result = db.execute(text("DESCRIBE account")) + columns = [row[0] for row in result.fetchall()] + print(f"Columns in 'account': {columns}") + except Exception as e: + print(f"Error describing 'account': {e}") + +if __name__ == "__main__": + check_columns() diff --git a/tests/check_site_schema.py b/tests/check_site_schema.py new file mode 100644 index 0000000..c6bd87b --- /dev/null +++ b/tests/check_site_schema.py @@ -0,0 +1,19 @@ +from app.db_sql import db +from sqlalchemy import text + +def check_schema(name): + print(f"--- Schema for {name} ---") + try: + result = db.execute(text(f"DESCRIBE `{name}`")) + for row in result.fetchall(): + print(f" {row[0]} ({row[1]})") + except Exception as e: + print(f" Error: {e}") + print("-" * 30) + +if __name__ == "__main__": + # Check tables + check_schema("site") + check_schema("site_domain") + # Check views used in CRUD V3 + check_schema("v_site_domain") diff --git a/tests/check_site_schema_remote.py b/tests/check_site_schema_remote.py new file mode 100644 index 0000000..5d5f6c9 --- /dev/null +++ b/tests/check_site_schema_remote.py @@ -0,0 +1,25 @@ +import requests +import json + +BASE_URL = "https://dev-api.oneskyit.com/v3/crud" +ACCOUNT_ID = "nqOzejLCDXM" + +def get_headers(): + return {"X-Account-ID": ACCOUNT_ID} + +def check_schema(obj): + print(f"--- Schema for {obj} ---") + url = f"{BASE_URL}/{obj}/schema" + r = requests.get(url, headers=get_headers()) + if r.status_code == 200: + data = r.json()['data'] + cols = [c['field'] for col in [data['database']['columns']] for c in col] + print(f"Columns: {cols}") + fields = list(data['model']['fields'].keys()) + print(f"Pydantic Fields: {fields}") + else: + print(f"Error {r.status_code}: {r.text}") + +if __name__ == "__main__": + check_schema("site") + check_schema("site_domain") diff --git a/tests/diagnose_boot.py b/tests/diagnose_boot.py new file mode 100644 index 0000000..bb26a1c --- /dev/null +++ b/tests/diagnose_boot.py @@ -0,0 +1,54 @@ +import sys +import os + +print("--- Starting Import Diagnosis ---") + +try: + print("1. Importing app.config...") + from app import config + print(f" Success. Settings found: {hasattr(config, 'settings')}") +except Exception as e: + print(f" FAILED: {e}") + sys.exit(1) + +try: + print("2. Importing app.log...") + import app.log + print(" Success.") +except Exception as e: + print(f" FAILED: {e}") + import traceback + traceback.print_exc() + sys.exit(1) + +try: + print("3. Importing app.db_connection...") + import app.db_connection + print(" Success.") +except Exception as e: + print(f" FAILED: {e}") + import traceback + traceback.print_exc() + sys.exit(1) + +try: + print("4. Importing app.db_sql...") + import app.db_sql + print(" Success.") +except Exception as e: + print(f" FAILED: {e}") + import traceback + traceback.print_exc() + sys.exit(1) + +try: + print("5. Importing app.main...") + import app.main + print(" Success.") +except Exception as e: + print(f" FAILED: {e}") + import traceback + traceback.print_exc() + sys.exit(1) + +print("--- Diagnosis Complete: No top-level import errors found in local env ---") diff --git a/mcp_docker_explorer.py b/tests/mcp_docker_explorer.py similarity index 100% rename from mcp_docker_explorer.py rename to tests/mcp_docker_explorer.py diff --git a/tests/test_agent_bridge.py b/tests/test_agent_bridge.py new file mode 100644 index 0000000..e3547af --- /dev/null +++ b/tests/test_agent_bridge.py @@ -0,0 +1,55 @@ +import requests +import json + +# Configuration +BASE_URL = "https://dev-api.oneskyit.com/agent" + +def get_headers(): + headers = { + "Content-Type": "application/json", + "X-No-Account-ID": "testing-bypass" + } + return headers + +def test_endpoint(method, path, description, params=None): + """ + Helper to run a test and print results. + """ + print(f"--- Testing: {description} ---") + url = f"{BASE_URL}{path}" + + request_headers = get_headers() + + try: + if method == "GET": + response = requests.get(url, headers=request_headers, params=params) + + print(f"URL: {response.url}") + print(f"Status Code: {response.status_code}") + + data = response.json() + + # Check if the result is a list or single object + result_data = data.get('data') + if isinstance(result_data, dict): + print(f"Result Data: {json.dumps(result_data, indent=2)}") + else: + print(f"Result Data (truncated): {str(result_data)[:200]}...") + + if response.status_code != 200: + print(f"Error Message: {data.get('status_message')}") + + except Exception as e: + print(f"Error during test: {e}") + print("-" * 40 + "\n") + +if __name__ == "__main__": + print(f"Starting Aether Agent Bridge Tests against {BASE_URL}\n") + + # 1. Get Status + test_endpoint("GET", "/status", "Get Container Status") + + # 2. Get Logs (last 5 lines) + test_endpoint("GET", "/logs", "Get Latest Logs", params={"lines": 5}) + + print("Tests Complete.") diff --git a/tests/test_site_lookup.py b/tests/test_site_lookup.py new file mode 100644 index 0000000..61c6ab9 --- /dev/null +++ b/tests/test_site_lookup.py @@ -0,0 +1,47 @@ +import requests +import json +import sys + +# Configuration +BASE_URL = "https://dev-api.oneskyit.com/v3/crud" +FQDN_TO_TEST = "dev-app.oneskyit.com" # Example domain that should exist + +def test_site_domain_lookup(): + print(f"--- Testing Public Site Domain Lookup: {FQDN_TO_TEST} ---") + url = f"{BASE_URL}/site_domain/search" + + # Payload: Search for FQDN, explicitly WITHOUT authentication + # The frontend typically sends this to bootstrap + query = { + "and": [{"field": "fqdn", "op": "eq", "value": FQDN_TO_TEST}] + } + + # NO AUTH HEADERS (Simulating unauthenticated bootstrapping) + headers = { + "Content-Type": "application/json" + } + + try: + response = requests.post(url, headers=headers, json=query) + print(f"URL: {response.url}") + print(f"Status Code: {response.status_code}") + + data = response.json() + + if response.status_code == 200: + result_data = data.get('data') + if isinstance(result_data, list): + print(f"Success! Found {len(result_data)} records.") + if len(result_data) > 0: + print(f"Data: {json.dumps(result_data[0], indent=2)}") + else: + print(f"Unexpected data format: {type(result_data)}") + else: + print(f"Failure (Expected before fix). Message: {data.get('status_message')}") + + except Exception as e: + print(f"Error during test: {e}") + print("-" * 40 + "\n") + +if __name__ == "__main__": + test_site_domain_lookup() diff --git a/tests/test_v3_accounts.py b/tests/test_v3_accounts.py new file mode 100644 index 0000000..31f6fc1 --- /dev/null +++ b/tests/test_v3_accounts.py @@ -0,0 +1,67 @@ +import requests +import json +import sys + +# Configuration +BASE_URL = "https://dev-api.oneskyit.com/v3/crud" + +# --- AUTHENTICATION CONFIG --- +ACCOUNT_ID = "nqOzejLCDXM" # Legacy Header Fallback + +def get_headers(): + headers = { + "Content-Type": "application/json", + "X-Account-ID": ACCOUNT_ID + } + return headers + +def test_endpoint(method, path, description, query=None, params=None): + """ + Helper to run a test and print results. + """ + print(f"--- Testing: {description} ---") + url = f"{BASE_URL}{path}" + + request_headers = get_headers() + + try: + if method == "GET": + response = requests.get(url, headers=request_headers, params=params) + elif method == "POST": + response = requests.post(url, headers=request_headers, json=query, params=params) + + print(f"URL: {response.url}") + print(f"Status Code: {response.status_code}") + + data = response.json() + + # Check if the result is a list or single object + result_data = data.get('data') + if isinstance(result_data, list): + print(f"Result Count: {len(result_data)}") + if len(result_data) > 0: + print(f"First Item Example: {json.dumps(result_data[0], indent=2)}") + else: + print(f"Result Type: {type(result_data)}") + print(f"Result Data: {json.dumps(result_data, indent=2)}") + + if response.status_code != 200: + print(f"Error Message: {data.get('status_message')}") + + except Exception as e: + print(f"Error during test: {e}") + print("-" * 40 + "\n") + +if __name__ == "__main__": + print(f"Starting Aether V3 Account Tests against {BASE_URL}\n") + + # 1. List Accounts + test_endpoint("GET", "/account/", "List Accounts (GET)") + + # 2. Search Accounts (Full Text) + test_endpoint("POST", "/account/search", "Search Accounts (POST - All)", query={"q": "%"}) + + # 3. Search Accounts (Specific Name) + test_endpoint("POST", "/account/search", "Search Accounts (POST - Specific)", query={"and": [{"field": "name", "op": "icontains", "value": "Sky"}]}) + + print("Tests Complete.") diff --git a/tests/test_v3_auth_isolation.py b/tests/test_v3_auth_isolation.py new file mode 100644 index 0000000..8ec629e --- /dev/null +++ b/tests/test_v3_auth_isolation.py @@ -0,0 +1,50 @@ + +import sys +import os +from fastapi.testclient import TestClient + +# Add the project root to sys.path so we can import 'app' +sys.path.append(os.getcwd()) + +from app.main import app + +client = TestClient(app) + +def test_site_domain_unauthenticated_search(): + """Test that searching site_domain works without authentication.""" + print("Testing unauthenticated site_domain search...") + # Using a simple search query that would typically be used to resolve FQDN + search_payload = { + "and_filters": [ + {"field": "fqdn", "op": "eq", "value": "aether.osit.dev"} + ] + } + response = client.post("/v3/crud/site_domain/search", json=search_payload) + print(f"Response Status: {response.status_code}") + print(f"Response Body: {response.json()}") + + # We expect 200 OK (even if empty results, the point is it's not 403) + assert response.status_code == 200 + assert response.json()["status"] == "success" + +def test_account_unauthenticated_search_blocked(): + """Test that searching other objects (e.g., account) is blocked without authentication.""" + print("\nTesting unauthenticated account search (should be blocked)...") + search_payload = { + "and_filters": [] + } + response = client.post("/v3/crud/account/search", json=search_payload) + print(f"Response Status: {response.status_code}") + + # We expect 403 Forbidden + assert response.status_code == 403 + assert "Authentication required" in response.json()["status_message"] + +if __name__ == "__main__": + try: + test_site_domain_unauthenticated_search() + test_account_unauthenticated_search_blocked() + print("\nSUCCESS: V3 Auth Isolation bypass for site_domain is working correctly.") + except Exception as e: + print(f"\nFAILURE: {e}") + sys.exit(1) diff --git a/tests/test_v3_extra.py b/tests/test_v3_extra.py new file mode 100644 index 0000000..838c650 --- /dev/null +++ b/tests/test_v3_extra.py @@ -0,0 +1,55 @@ +import requests +import json + +# Configuration +BASE_URL = "https://dev-api.oneskyit.com/v3/crud" +ACCOUNT_ID = "nqOzejLCDXM" # Legacy Header Fallback + +def get_headers(include_account=True): + headers = { + "Content-Type": "application/json" + } + if include_account: + headers["X-Account-ID"] = ACCOUNT_ID + return headers + +def test_endpoint(path, description, include_account=True): + print(f"--- Testing: {description} ---") + url = f"{BASE_URL}{path}" + try: + response = requests.get(url, headers=get_headers(include_account)) + print(f"URL: {response.url}") + print(f"Status Code: {response.status_code}") + + data = response.json() + result_data = data.get('data') + + if isinstance(result_data, list): + print(f"Result Count: {len(result_data)}") + if len(result_data) > 0: + print(f"First Item: {json.dumps(result_data[0], indent=2)[:300]}...") + else: + print(f"Result Data: {json.dumps(result_data, indent=2)}") + + if response.status_code != 200: + print(f"Error Message: {data.get('status_message')}") + + except Exception as e: + print(f"Error during test: {e}") + print("-" * 40 + "\n") + +if __name__ == "__main__": + # Test Users + test_endpoint("/user/", "List Users (Default filter)") + test_endpoint("/user/?enabled=all&hidden=all", "List Users (Bypass filter)") + + # Test Sites + test_endpoint(f"/account/{ACCOUNT_ID}/site/", "List Sites (Account Filter)") + + # Test Site Domains + test_endpoint("/site_domain/", "List Site Domains (Default filter)") + test_endpoint("/site_domain/?enabled=all&hidden=all", "List Site Domains (Bypass filter)") + + # Test Legacy Site Domain Lookup (Initial frontend request) + # This route is in api_crud.py (v1) and needs to work without an account ID header + test_endpoint("/../../crud/site/domain/scott.localhost:5173?use_alt_table=true&use_alt_base=true", "Legacy Site Domain Lookup", include_account=False) diff --git a/tests/test_v3_schema.py b/tests/test_v3_schema.py new file mode 100644 index 0000000..4aa8171 --- /dev/null +++ b/tests/test_v3_schema.py @@ -0,0 +1,49 @@ +import requests +import json + +# Configuration +BASE_URL = "https://dev-api.oneskyit.com/v3/crud" +ACCOUNT_ID = "nqOzejLCDXM" # Legacy Header Fallback + +def get_headers(): + return { + "Content-Type": "application/json", + "X-Account-ID": ACCOUNT_ID + } + +def test_schema(obj_type): + print(f"--- Testing Schema: {obj_type} ---") + url = f"{BASE_URL}/{obj_type}/schema" + try: + response = requests.get(url, headers=get_headers()) + print(f"URL: {response.url}") + print(f"Status Code: {response.status_code}") + + data = response.json() + if response.status_code == 200: + print(f"Object Type: {data['data']['object_type']}") + print(f"Database Table: {data['data']['database']['table_name']}") + print(f"Column Count: {len(data['data']['database']['columns'])}") + print(f"Model Name: {data['data']['model']['name']}") + print(f"Model Field Count: {len(data['data']['model']['fields'])}") + + # Print a few columns and fields as example + print("\nExample Columns:") + for col in data['data']['database']['columns'][:3]: + print(f" - {col['field']} ({col['type']})") + + print("\nExample Model Fields:") + fields = list(data['data']['model']['fields'].keys()) + for field in fields[:3]: + f_info = data['data']['model']['fields'][field] + print(f" - {field} (alias: {f_info['alias']}, type: {f_info['type']})") + else: + print(f"Error: {data.get('status_message')}") + + except Exception as e: + print(f"Error during test: {e}") + print("-" * 40 + "\n") + +if __name__ == "__main__": + test_schema("account") + test_schema("event_badge") diff --git a/admin/development/test_v3_search.py b/tests/test_v3_search.py similarity index 100% rename from admin/development/test_v3_search.py rename to tests/test_v3_search.py