Upgrade/phase-1-python-312#101
Merged
Merged
Conversation
- Update runtime.txt to python-3.12.2
- Update non-breaking dependencies:
- gunicorn 20.1.0 → 22.0.0 (security patches)
- firebase-admin 3.2.1 → 6.5.0
- pymysql 1.0.2 → 1.1.1
- python-dotenv 0.15.0 → 1.0.0
- Add google-auth>=2.22.0, requests>=2.31.0, cryptography>=41.0.0
- Fix Python 3.12 deprecations:
- Replace datetime.utcnow() with datetime.now(timezone.utc) in my_oauth.py
- Remove .decode('utf-8') from jwt.encode() in ReportState.py (google-auth 2.x returns str)
- Replace all print() statements with logging module:
- action_devices.py: 16 print() → logger calls
- app.py: 12 print() → logger calls
- notifications.py: 5 print() → logger calls
- generate_service_account_file.py: 4 print() → logger calls
- ReportState.py: 3 print() → logger calls
Tests: All 6 tests pass ✓
Replace all 11 print() statements with structured logger.debug() calls: - get_current_user(): log user lookup results - load_client(): log client ID and lookup results - load_grant(): log grant request - save_grant(): log grant creation - load_token(): log token request - save_token(): log token creation and existing tokens Maintains consistency with logging infrastructure added in earlier Phase 1 commits. All tests passing (10/10).
Contributor
Reviewer's GuideReplaces ad-hoc print-based debugging with structured logging throughout the app and OAuth code, adds runtime/tests around auth and device endpoints, modernizes JWT/report-state handling, and updates several dependencies for Python 3.12 compatibility and security. File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
Contributor
Summary of ChangesHello, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This pull request primarily focuses on upgrading the project's Python environment to version 3.12.2 and modernizing its logging practices. It systematically replaces all direct Highlights
Changelog
Activity
Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for GitHub and other Google products, sign up here. You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension. Footnotes
|
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Hey - I've found 4 issues, and left some high level feedback:
- In
my_oauth.py, prefer logger format arguments (e.g.,logger.debug('Current user: %s', user)) instead of f-strings to avoid doing string interpolation when the log level is disabled and to keep logging style consistent with the rest of the changes. - The new use of
datetime.now(timezone.utc).replace(tzinfo=None)insave_grantandsave_tokenis a bit convoluted; if you want a naive UTC timestamp similar to the previous behavior, consider keepingdatetime.utcnow()for clarity or making the timestamps fully timezone-aware and updating the DB model accordingly.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In `my_oauth.py`, prefer logger format arguments (e.g., `logger.debug('Current user: %s', user)`) instead of f-strings to avoid doing string interpolation when the log level is disabled and to keep logging style consistent with the rest of the changes.
- The new use of `datetime.now(timezone.utc).replace(tzinfo=None)` in `save_grant` and `save_token` is a bit convoluted; if you want a naive UTC timestamp similar to the previous behavior, consider keeping `datetime.utcnow()` for clarity or making the timestamps fully timezone-aware and updating the DB model accordingly.
## Individual Comments
### Comment 1
<location path="ReportState.py" line_range="55-56" />
<code_context>
- print('By ReportState')
- signed_jwt = generate_jwt(service_account).decode("utf-8") # Decode
+ logger.info('By ReportState')
+ # google-auth >= 2.x returns a str; no .decode() needed
+ signed_jwt = generate_jwt(service_account)
access_token = get_access_token(signed_jwt)
success = report_state(access_token, report_state_file)
</code_context>
<issue_to_address>
**issue (bug_risk):** The removal of `.decode("utf-8")` may break compatibility if `jwt.encode` still returns bytes.
`google.auth.jwt.encode` has historically returned `bytes`, and some environments may still see that behavior. If `generate_jwt` returns `bytes`, dropping `.decode("utf-8")` will make `signed_jwt` a `bytes` value and can break `get_access_token` or any consumers that expect a string (e.g. headers/JSON). Please either keep the explicit `.decode("utf-8")` or guarantee (and document) that `generate_jwt` always returns a `str`, so behavior doesn’t silently depend on the installed `google-auth` version.
</issue_to_address>
### Comment 2
<location path="tests/test_app.py" line_range="110-115" />
<code_context>
+ db.session.delete(result)
+ db.session.commit()
+
+ def test_signup_and_login(self):
+ resp = self.client.post('/signup', data={'email': 'a@b.com', 'name': 'A', 'password': 'pass'})
+ self.assertIn(resp.status_code, (302, 200))
+ resp2 = self.client.post('/login', data={'email': 'a@b.com', 'password': 'pass'}, follow_redirects=True)
+ self.assertEqual(resp2.status_code, 200)
+ self.assertIn('Profile', resp2.get_data(as_text=True))
+
+
</code_context>
<issue_to_address>
**suggestion (testing):** Add negative-path auth tests (e.g., bad password, duplicate email, missing fields) to better cover the signup/login flows.
Currently we only cover the happy path for signup and login. Please add tests for at least:
- Signup with an email that already exists (should fail with a clear error).
- Login with an incorrect password (should not authenticate and should return the expected status/flash).
Optionally, also cover signup with missing fields to assert validation behavior and error responses.
Suggested implementation:
```python
def test_signup_and_login(self):
resp = self.client.post(
'/signup',
data={'email': 'a@b.com', 'name': 'A', 'password': 'pass'},
)
self.assertIn(resp.status_code, (302, 200))
resp2 = self.client.post(
'/login',
data={'email': 'a@b.com', 'password': 'pass'},
follow_redirects=True,
)
self.assertEqual(resp2.status_code, 200)
self.assertIn('Profile', resp2.get_data(as_text=True))
def test_signup_with_duplicate_email_fails(self):
# First signup should succeed
first_resp = self.client.post(
'/signup',
data={'email': 'a@b.com', 'name': 'A', 'password': 'pass'},
follow_redirects=True,
)
self.assertIn(first_resp.status_code, (200, 302))
# Second signup with the same email should fail with a clear error
second_resp = self.client.post(
'/signup',
data={'email': 'a@b.com', 'name': 'A2', 'password': 'pass2'},
follow_redirects=True,
)
# Request should be handled but not treated as a successful signup
self.assertEqual(second_resp.status_code, 200)
body = second_resp.get_data(as_text=True)
self.assertIn('Email already', body)
def test_login_with_incorrect_password_fails(self):
# Create a valid user
signup_resp = self.client.post(
'/signup',
data={'email': 'a@b.com', 'name': 'A', 'password': 'correct-pass'},
follow_redirects=True,
)
self.assertIn(signup_resp.status_code, (200, 302))
# Attempt to login with an incorrect password
login_resp = self.client.post(
'/login',
data={'email': 'a@b.com', 'password': 'wrong-pass'},
follow_redirects=True,
)
# User should not be authenticated; login page should be re-rendered with an error
self.assertEqual(login_resp.status_code, 200)
body = login_resp.get_data(as_text=True)
self.assertIn('Invalid email or password', body)
def test_signup_missing_required_fields_shows_errors(self):
# Missing email
missing_email_resp = self.client.post(
'/signup',
data={'email': '', 'name': 'A', 'password': 'pass'},
follow_redirects=True,
)
self.assertEqual(missing_email_resp.status_code, 200)
body = missing_email_resp.get_data(as_text=True)
self.assertIn('Email', body)
# Missing password
missing_password_resp = self.client.post(
'/signup',
data={'email': 'a@b.com', 'name': 'A', 'password': ''},
follow_redirects=True,
)
self.assertEqual(missing_password_resp.status_code, 200)
body = missing_password_resp.get_data(as_text=True)
self.assertIn('Password', body)
```
You may need to adjust:
1. The exact error message substrings (`'Email already'`, `'Invalid email or password'`, `'Email'`, `'Password'`) to match the actual flash/validation messages your views render.
2. The expected status codes if your signup/login endpoints return something other than 200 on validation failure (e.g., 400).
3. If your teardown only deletes a single email and you introduce other test emails elsewhere, extend the teardown to clean up any additional test users you create.
</issue_to_address>
### Comment 3
<location path="tests/test_app.py" line_range="123-125" />
<code_context>
+ flask_app.config.update(TESTING=True)
+ self.client = flask_app.test_client()
+
+ def test_devices_requires_login(self):
+ resp = self.client.get('/devices')
+ self.assertEqual(resp.status_code, 302)
+
+ def test_smarthome_sync(self):
</code_context>
<issue_to_address>
**suggestion (testing):** Consider asserting on the redirect target and guarding this test with FULL_FEATURES so it behaves predictably across environments.
Right now this only checks for a 302; please also assert the redirect `Location` (e.g., login page) to directly confirm the auth requirement for `/devices`. In addition, consider `@unittest.skipUnless(FULL_FEATURES, "Devices route requires full feature set")` so this test doesn’t fail in minimal configurations where `/devices` behaves differently.
Suggested implementation:
```python
@unittest.skipUnless(FULL_FEATURES, "Devices route requires full feature set")
def test_devices_requires_login(self):
resp = self.client.get('/devices')
self.assertEqual(resp.status_code, 302)
location = resp.headers.get('Location', '')
self.assertIn('/login', location)
```
If `FULL_FEATURES` is not already imported or defined in `tests/test_app.py`, you will also need to:
1. Import or reference it consistently with the rest of the test suite, for example:
- `from app import FULL_FEATURES` or
- `from your_module import FULL_FEATURES`
2. Ensure `unittest` is imported at the top of the file (it likely already is since `unittest.TestCase` is used).
</issue_to_address>
### Comment 4
<location path="tests/test_app.py" line_range="139-148" />
<code_context>
+ def setUp(self):
+ flask_app.config.update(TESTING=True)
+
+ def test_onSync_and_helpers(self):
+ with flask_app.app_context():
+ from action_devices import onSync, onQuery, onExecute, request_sync
+ sync = onSync()
+ self.assertIn('agentUserId', sync)
+ q = onQuery({'inputs': [{'payload': {'devices': [{'id': 'nonexistent'}]}}]})
+ self.assertIn('devices', q)
+ exec_resp = onExecute({'inputs': [{'payload': {'devices': [{'id': 'nonexistent'}]},
+ 'commands': [{'execution': [{'command': 'action.devices.commands.OnOff',
+ 'params': {'on': True}}]}]}]})
+ self.assertIn('commands', exec_resp)
+ self.assertEqual(exec_resp['commands'][0]['status'], 'ERROR')
+ self.assertFalse(request_sync('', ''))
</code_context>
<issue_to_address>
**suggestion (testing):** Avoid real external calls in `request_sync` and add more granular assertions for the action_devices helpers.
The test currently risks making a real HTTP call via `request_sync('', '')`. Please mock `requests.post` (e.g., with `unittest.mock.patch`) so it stays a true unit test without external dependencies.
Also consider splitting this into smaller tests per helper (`onSync`, `onQuery`, `onExecute`, `request_sync`) and add targeted assertions, such as checking specific error statuses/errorCodes for malformed `onExecute` commands and covering both existing and non-existing device IDs in `onQuery`. This will give more precise coverage of error handling and helper behavior.
Suggested implementation:
```python
class ActionDevicesUnitTests(unittest.TestCase):
def setUp(self):
flask_app.config.update(TESTING=True)
# Use a real app context so action_devices helpers can access Flask config/state.
self.app_context = flask_app.app_context()
self.app_context.push()
def tearDown(self):
# Ensure app context is cleaned up after each test.
self.app_context.pop()
def test_onSync_returns_agent_user_id(self):
from action_devices import onSync
sync = onSync()
self.assertIsInstance(sync, dict)
self.assertIn('agentUserId', sync)
def test_onQuery_handles_existing_and_nonexistent_devices(self):
from action_devices import onQuery
# Use a mix of existing and non-existing IDs to exercise error handling.
# NOTE: adjust `existing_device_id` to match a real device ID from action_devices.
existing_device_id = 'existing-device-id'
nonexistent_device_id = 'nonexistent-device-id'
query_payload = {
'inputs': [
{
'payload': {
'devices': [
{'id': existing_device_id},
{'id': nonexistent_device_id},
]
}
}
]
}
q = onQuery(query_payload)
self.assertIn('devices', q)
devices = q['devices']
self.assertIsInstance(devices, dict)
# Existing device should be present and not in an error state (details depend on implementation).
self.assertIn(existing_device_id, devices)
# Non-existent device should be present with an error status / errorCode if supported.
self.assertIn(nonexistent_device_id, devices)
nonexistent_info = devices[nonexistent_device_id]
self.assertIsInstance(nonexistent_info, dict)
if 'status' in nonexistent_info:
self.assertEqual(nonexistent_info['status'], 'ERROR')
if 'errorCode' in nonexistent_info:
# Typical GSH error for missing device; adjust if implementation differs.
self.assertIn(nonexistent_info['errorCode'], ('deviceNotFound', 'notFound', 'unknownError'))
def test_onExecute_returns_error_for_invalid_device_or_command(self):
from action_devices import onExecute
execute_payload = {
'inputs': [
{
'payload': {
'devices': [{'id': 'nonexistent-device-id'}],
},
'commands': [
{
'execution': [
{
'command': 'action.devices.commands.OnOff',
'params': {'on': True},
}
]
}
],
}
]
}
exec_resp = onExecute(execute_payload)
self.assertIn('commands', exec_resp)
commands = exec_resp['commands']
self.assertIsInstance(commands, list)
self.assertGreaterEqual(len(commands), 1)
cmd_result = commands[0]
self.assertIsInstance(cmd_result, dict)
self.assertEqual(cmd_result.get('status'), 'ERROR')
# Ensure errorCode is surfaced when available.
if 'errorCode' in cmd_result:
self.assertIsInstance(cmd_result['errorCode'], str)
self.assertTrue(cmd_result['errorCode'])
def test_request_sync_handles_successful_response(self):
from unittest.mock import patch, MagicMock
from action_devices import request_sync
with patch('action_devices.requests.post') as mock_post:
mock_response = MagicMock()
mock_response.status_code = 200
mock_post.return_value = mock_response
result = request_sync('test-agent-user', 'test-api-key')
# Should treat HTTP 200 as success.
self.assertTrue(result)
mock_post.assert_called_once()
args, kwargs = mock_post.call_args
self.assertIsInstance(args[0], str) # URL
self.assertIn('headers', kwargs)
self.assertIn('json', kwargs)
def test_request_sync_handles_error_response(self):
from unittest.mock import patch, MagicMock
from action_devices import request_sync
with patch('action_devices.requests.post') as mock_post:
mock_response = MagicMock()
mock_response.status_code = 500
mock_post.return_value = mock_response
result = request_sync('test-agent-user', 'test-api-key')
# Should not raise and should return False on non-200 responses.
self.assertFalse(result)
mock_post.assert_called_once()
```
1. The `existing_device_id` value in `test_onQuery_handles_existing_and_nonexistent_devices` must be updated to match a real device ID defined/known in `action_devices` (e.g. from a devices registry or fixture). The test currently assumes such an ID exists.
2. The structure of the `onQuery` and `onExecute` responses is inferred. If the actual implementation nests results under a `payload` key or uses a different shape, adjust the assertions accordingly (e.g., `q['payload']['devices']` instead of `q['devices']`).
3. The `request_sync` tests assume `request_sync` returns `True` on HTTP 200 and `False` on non-200 responses, and that `action_devices` imports `requests` as `import requests`. If `request_sync` uses a different success criterion or a different import path, update the patch target (`'action_devices.requests.post'`) and expected return values.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Comment on lines
+110
to
+115
| def test_signup_and_login(self): | ||
| resp = self.client.post('/signup', data={'email': 'a@b.com', 'name': 'A', 'password': 'pass'}) | ||
| self.assertIn(resp.status_code, (302, 200)) | ||
| resp2 = self.client.post('/login', data={'email': 'a@b.com', 'password': 'pass'}, follow_redirects=True) | ||
| self.assertEqual(resp2.status_code, 200) | ||
| self.assertIn('Profile', resp2.get_data(as_text=True)) |
Contributor
There was a problem hiding this comment.
suggestion (testing): Add negative-path auth tests (e.g., bad password, duplicate email, missing fields) to better cover the signup/login flows.
Currently we only cover the happy path for signup and login. Please add tests for at least:
- Signup with an email that already exists (should fail with a clear error).
- Login with an incorrect password (should not authenticate and should return the expected status/flash).
Optionally, also cover signup with missing fields to assert validation behavior and error responses.
Suggested implementation:
def test_signup_and_login(self):
resp = self.client.post(
'/signup',
data={'email': 'a@b.com', 'name': 'A', 'password': 'pass'},
)
self.assertIn(resp.status_code, (302, 200))
resp2 = self.client.post(
'/login',
data={'email': 'a@b.com', 'password': 'pass'},
follow_redirects=True,
)
self.assertEqual(resp2.status_code, 200)
self.assertIn('Profile', resp2.get_data(as_text=True))
def test_signup_with_duplicate_email_fails(self):
# First signup should succeed
first_resp = self.client.post(
'/signup',
data={'email': 'a@b.com', 'name': 'A', 'password': 'pass'},
follow_redirects=True,
)
self.assertIn(first_resp.status_code, (200, 302))
# Second signup with the same email should fail with a clear error
second_resp = self.client.post(
'/signup',
data={'email': 'a@b.com', 'name': 'A2', 'password': 'pass2'},
follow_redirects=True,
)
# Request should be handled but not treated as a successful signup
self.assertEqual(second_resp.status_code, 200)
body = second_resp.get_data(as_text=True)
self.assertIn('Email already', body)
def test_login_with_incorrect_password_fails(self):
# Create a valid user
signup_resp = self.client.post(
'/signup',
data={'email': 'a@b.com', 'name': 'A', 'password': 'correct-pass'},
follow_redirects=True,
)
self.assertIn(signup_resp.status_code, (200, 302))
# Attempt to login with an incorrect password
login_resp = self.client.post(
'/login',
data={'email': 'a@b.com', 'password': 'wrong-pass'},
follow_redirects=True,
)
# User should not be authenticated; login page should be re-rendered with an error
self.assertEqual(login_resp.status_code, 200)
body = login_resp.get_data(as_text=True)
self.assertIn('Invalid email or password', body)
def test_signup_missing_required_fields_shows_errors(self):
# Missing email
missing_email_resp = self.client.post(
'/signup',
data={'email': '', 'name': 'A', 'password': 'pass'},
follow_redirects=True,
)
self.assertEqual(missing_email_resp.status_code, 200)
body = missing_email_resp.get_data(as_text=True)
self.assertIn('Email', body)
# Missing password
missing_password_resp = self.client.post(
'/signup',
data={'email': 'a@b.com', 'name': 'A', 'password': ''},
follow_redirects=True,
)
self.assertEqual(missing_password_resp.status_code, 200)
body = missing_password_resp.get_data(as_text=True)
self.assertIn('Password', body)
You may need to adjust:
- The exact error message substrings (
'Email already','Invalid email or password','Email','Password') to match the actual flash/validation messages your views render. - The expected status codes if your signup/login endpoints return something other than 200 on validation failure (e.g., 400).
- If your teardown only deletes a single email and you introduce other test emails elsewhere, extend the teardown to clean up any additional test users you create.
Comment on lines
+123
to
+125
| def test_devices_requires_login(self): | ||
| resp = self.client.get('/devices') | ||
| self.assertEqual(resp.status_code, 302) |
Contributor
There was a problem hiding this comment.
suggestion (testing): Consider asserting on the redirect target and guarding this test with FULL_FEATURES so it behaves predictably across environments.
Right now this only checks for a 302; please also assert the redirect Location (e.g., login page) to directly confirm the auth requirement for /devices. In addition, consider @unittest.skipUnless(FULL_FEATURES, "Devices route requires full feature set") so this test doesn’t fail in minimal configurations where /devices behaves differently.
Suggested implementation:
@unittest.skipUnless(FULL_FEATURES, "Devices route requires full feature set")
def test_devices_requires_login(self):
resp = self.client.get('/devices')
self.assertEqual(resp.status_code, 302)
location = resp.headers.get('Location', '')
self.assertIn('/login', location)If FULL_FEATURES is not already imported or defined in tests/test_app.py, you will also need to:
- Import or reference it consistently with the rest of the test suite, for example:
from app import FULL_FEATURESorfrom your_module import FULL_FEATURES
- Ensure
unittestis imported at the top of the file (it likely already is sinceunittest.TestCaseis used).
Comment on lines
+139
to
+148
| def test_onSync_and_helpers(self): | ||
| with flask_app.app_context(): | ||
| from action_devices import onSync, onQuery, onExecute, request_sync | ||
| sync = onSync() | ||
| self.assertIn('agentUserId', sync) | ||
| q = onQuery({'inputs': [{'payload': {'devices': [{'id': 'nonexistent'}]}}]}) | ||
| self.assertIn('devices', q) | ||
| exec_resp = onExecute({'inputs': [{'payload': {'devices': [{'id': 'nonexistent'}]}, | ||
| 'commands': [{'execution': [{'command': 'action.devices.commands.OnOff', | ||
| 'params': {'on': True}}]}]}]}) |
Contributor
There was a problem hiding this comment.
suggestion (testing): Avoid real external calls in request_sync and add more granular assertions for the action_devices helpers.
The test currently risks making a real HTTP call via request_sync('', ''). Please mock requests.post (e.g., with unittest.mock.patch) so it stays a true unit test without external dependencies.
Also consider splitting this into smaller tests per helper (onSync, onQuery, onExecute, request_sync) and add targeted assertions, such as checking specific error statuses/errorCodes for malformed onExecute commands and covering both existing and non-existing device IDs in onQuery. This will give more precise coverage of error handling and helper behavior.
Suggested implementation:
class ActionDevicesUnitTests(unittest.TestCase):
def setUp(self):
flask_app.config.update(TESTING=True)
# Use a real app context so action_devices helpers can access Flask config/state.
self.app_context = flask_app.app_context()
self.app_context.push()
def tearDown(self):
# Ensure app context is cleaned up after each test.
self.app_context.pop()
def test_onSync_returns_agent_user_id(self):
from action_devices import onSync
sync = onSync()
self.assertIsInstance(sync, dict)
self.assertIn('agentUserId', sync)
def test_onQuery_handles_existing_and_nonexistent_devices(self):
from action_devices import onQuery
# Use a mix of existing and non-existing IDs to exercise error handling.
# NOTE: adjust `existing_device_id` to match a real device ID from action_devices.
existing_device_id = 'existing-device-id'
nonexistent_device_id = 'nonexistent-device-id'
query_payload = {
'inputs': [
{
'payload': {
'devices': [
{'id': existing_device_id},
{'id': nonexistent_device_id},
]
}
}
]
}
q = onQuery(query_payload)
self.assertIn('devices', q)
devices = q['devices']
self.assertIsInstance(devices, dict)
# Existing device should be present and not in an error state (details depend on implementation).
self.assertIn(existing_device_id, devices)
# Non-existent device should be present with an error status / errorCode if supported.
self.assertIn(nonexistent_device_id, devices)
nonexistent_info = devices[nonexistent_device_id]
self.assertIsInstance(nonexistent_info, dict)
if 'status' in nonexistent_info:
self.assertEqual(nonexistent_info['status'], 'ERROR')
if 'errorCode' in nonexistent_info:
# Typical GSH error for missing device; adjust if implementation differs.
self.assertIn(nonexistent_info['errorCode'], ('deviceNotFound', 'notFound', 'unknownError'))
def test_onExecute_returns_error_for_invalid_device_or_command(self):
from action_devices import onExecute
execute_payload = {
'inputs': [
{
'payload': {
'devices': [{'id': 'nonexistent-device-id'}],
},
'commands': [
{
'execution': [
{
'command': 'action.devices.commands.OnOff',
'params': {'on': True},
}
]
}
],
}
]
}
exec_resp = onExecute(execute_payload)
self.assertIn('commands', exec_resp)
commands = exec_resp['commands']
self.assertIsInstance(commands, list)
self.assertGreaterEqual(len(commands), 1)
cmd_result = commands[0]
self.assertIsInstance(cmd_result, dict)
self.assertEqual(cmd_result.get('status'), 'ERROR')
# Ensure errorCode is surfaced when available.
if 'errorCode' in cmd_result:
self.assertIsInstance(cmd_result['errorCode'], str)
self.assertTrue(cmd_result['errorCode'])
def test_request_sync_handles_successful_response(self):
from unittest.mock import patch, MagicMock
from action_devices import request_sync
with patch('action_devices.requests.post') as mock_post:
mock_response = MagicMock()
mock_response.status_code = 200
mock_post.return_value = mock_response
result = request_sync('test-agent-user', 'test-api-key')
# Should treat HTTP 200 as success.
self.assertTrue(result)
mock_post.assert_called_once()
args, kwargs = mock_post.call_args
self.assertIsInstance(args[0], str) # URL
self.assertIn('headers', kwargs)
self.assertIn('json', kwargs)
def test_request_sync_handles_error_response(self):
from unittest.mock import patch, MagicMock
from action_devices import request_sync
with patch('action_devices.requests.post') as mock_post:
mock_response = MagicMock()
mock_response.status_code = 500
mock_post.return_value = mock_response
result = request_sync('test-agent-user', 'test-api-key')
# Should not raise and should return False on non-200 responses.
self.assertFalse(result)
mock_post.assert_called_once()- The
existing_device_idvalue intest_onQuery_handles_existing_and_nonexistent_devicesmust be updated to match a real device ID defined/known inaction_devices(e.g. from a devices registry or fixture). The test currently assumes such an ID exists. - The structure of the
onQueryandonExecuteresponses is inferred. If the actual implementation nests results under apayloadkey or uses a different shape, adjust the assertions accordingly (e.g.,q['payload']['devices']instead ofq['devices']). - The
request_synctests assumerequest_syncreturnsTrueon HTTP 200 andFalseon non-200 responses, and thataction_devicesimportsrequestsasimport requests. Ifrequest_syncuses a different success criterion or a different import path, update the patch target ('action_devices.requests.post') and expected return values.
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Code Review
This pull request successfully upgrades the application for Python 3.12 compatibility and replaces all print statements with structured logging, significantly improving maintainability and debugging. However, a critical security vulnerability has been identified: an Insecure Direct Object Reference (IDOR) / Path Traversal vulnerability in how device IDs are used to access the Firebase database, and a complete lack of authentication on the /smarthome endpoint. These issues allow unauthorized access to the entire database and control over all connected devices, requiring immediate remediation. A minor suggestion for logging best practices for performance is also noted.
…ask-Login integration
…r optional features
…ogging; remove feature flags
…andling and test isolation
…d MQTT initialization error handling
DaTiC0
commented
Mar 9, 2026
…unction for schema setup and update CLI command
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Replace all 11 print() statements with structured logger.debug() calls:
Maintains consistency with logging infrastructure added in earlier Phase 1 commits. All tests passing (10/10).
Summary
Validation
make test)make health)AI-Assisted Review (if applicable)
Risk & Rollback
Summary by Sourcery
Replace ad-hoc printing with structured logging, modernize OAuth and Google Home integrations, expand test coverage, and update dependencies for Python 3.12 compatibility.
Enhancements:
Build:
Tests: