Privacy consultation bed #2506
Privacy consultation bed #2506
Conversation
Codecov ReportAttention: Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## develop #2506 +/- ##
===========================================
+ Coverage 70.03% 70.06% +0.02%
===========================================
Files 198 199 +1
Lines 11674 11779 +105
Branches 1607 1626 +19
===========================================
+ Hits 8176 8253 +77
- Misses 3107 3132 +25
- Partials 391 394 +3 ☔ View full report in Codecov by Sentry. 🚨 Try these New Features:
|
| send_webpush(username=current_user, message=message) | ||
|
|
||
| def lock_camera(self) -> bool: | ||
| current_user = cache.get(self.current_user_cache_key) |
There was a problem hiding this comment.
This is not how we implement locks in care, refer to other lock implementations
There was a problem hiding this comment.
@vigneshhari This implementation was based on the existing implementation of locks in discharge summary (https://github.com/ohcnetwork/care/blob/develop/care/facility/utils/reports/discharge_summary.py#L40)
|
Why are there no tests for this change? |
|
Temporarily blocking this PR, will rework it once the camera plug is created. |
📝 WalkthroughWalkthroughThe pull request introduces several enhancements across multiple files, primarily focusing on error handling and access control. Key changes include the addition of the Changes
Possibly related PRs
Suggested labels
Suggested reviewers
Warning Rate limit exceeded@khavinshankar has exceeded the limit for the number of commits or files that can be reviewed per hour. Please wait 28 minutes and 9 seconds before requesting another review. ⌛ How to resolve this issue?After the wait time has elapsed, a review can be triggered using the We recommend that you space out your commits to avoid hitting the rate limit. 🚦 How do rate limits work?CodeRabbit enforces hourly rate limits for each developer per organization. Our paid plans have higher rate limits than the trial, open-source and free plans. In all cases, we re-allow further reviews after a brief timeout. Please see our FAQ for further information. Thank you for using CodeRabbit. We offer it for free to the OSS community and would appreciate your support in helping us grow. If you find it useful, would you consider giving us a shout-out on your favorite social media? 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 14
🧹 Outside diff range and nitpick comments (16)
care/utils/assetintegration/hl7monitor.py (2)
Line range hint
23-39: Error handling seems a bit... optimisticThe API calls (
api_getandapi_post) lack proper error handling. In a medical context, we should probably be a tiny bit more careful about handling failures, don't you think?Consider wrapping the API calls in try-catch blocks:
def handle_action(self, action, user): action_type = action["type"] if action_type == self.HL7MonitorActions.GET_VITALS.value: request_params = {"device_id": self.host} - return self.api_get(self.get_url("vitals"), request_params) + try: + return self.api_get(self.get_url("vitals"), request_params) + except Exception as e: + raise ValidationError({"error": f"Failed to fetch vitals: {str(e)}"}) if action_type == self.HL7MonitorActions.GET_STREAM_TOKEN.value: - return self.api_post( - self.get_url("api/stream/getToken/vitals"), - { - "asset_id": self.id, - "ip": self.host, - }, - ) + try: + return self.api_post( + self.get_url("api/stream/getToken/vitals"), + { + "asset_id": self.id, + "ip": self.host, + }, + ) + except Exception as e: + raise ValidationError({"error": f"Failed to get stream token: {str(e)}"})
Line range hint
23-39: Medical data access without authorization checks? How... interesting.Given that this handles sensitive medical data (vitals), we should probably add some authorization checks using the provided user context.
Consider adding permission checks before processing the actions:
def handle_action(self, action, user): action_type = action["type"] + if not user.has_perm('assets.can_access_vitals'): + raise PermissionDenied("User does not have permission to access vitals data") + if action_type == self.HL7MonitorActions.GET_VITALS.value: request_params = {"device_id": self.host}Also, consider passing the user context to the API endpoints for audit trails:
request_params = {"device_id": self.host} + request_params["user_id"] = user.id return self.api_get(self.get_url("vitals"), request_params)care/utils/assetintegration/ventilator.py (2)
Line range hint
23-39: Missing permission checks for ventilator actionsGiven that we're adding user context, we should probably verify if the user has appropriate permissions to perform these ventilator-related actions. You know, just to make sure we're not letting anyone play around with medical equipment.
Consider adding permission checks before executing the actions. Here's a suggested implementation:
def handle_action(self, action, user): action_type = action["type"] + + if not user.has_perm('ventilator.can_access_vitals'): + raise PermissionDenied("User does not have permission to access ventilator data") if action_type == self.VentilatorActions.GET_VITALS.value: request_params = {"device_id": self.host}
Line range hint
23-39: Security considerations for ventilator data accessA few security-related suggestions that might be worth considering (when you have the time, of course):
- Implement rate limiting for API calls
- Add validation for the response data
- Include proper error handling for API failures
Would you like me to provide a more detailed implementation for any of these suggestions?
care/utils/assetintegration/base.py (1)
22-23: Would it kill you to add some documentation and type hints? 😊The new
userparameter needs proper documentation and type hints for better maintainability. Consider:- def handle_action(self, action, user): - pass + def handle_action(self, action: str, user: 'User') -> None: + """Handle asset-specific actions with user context. + + Args: + action: The action to perform + user: The user performing the action + """ + passcare/facility/models/bed.py (1)
95-95: Would it kill you to add some documentation?The new
is_privacy_enabledfield lacks documentation explaining its purpose, implications, and usage guidelines. While I'm sure it makes perfect sense to you, others might appreciate knowing what this actually does.Add a docstring explaining:
class ConsultationBed(BaseModel): consultation = models.ForeignKey( PatientConsultation, on_delete=models.PROTECT, null=False, blank=False ) bed = models.ForeignKey(Bed, on_delete=models.PROTECT, null=False, blank=False) start_date = models.DateTimeField(null=False, blank=False) end_date = models.DateTimeField(null=True, blank=True, default=None) + # Controls privacy settings for consultation bed + # When enabled, restricts access to consultation data and associated assets + # Note: This setting affects camera feeds and monitoring capabilities is_privacy_enabled = models.BooleanField(default=False)care/facility/tests/test_asset_usage_manager.py (4)
1-10: Oh, I see we're not using type hints... that's... interesting.While the code works perfectly fine without them, adding type hints would make it slightly more maintainable and somewhat easier to understand. Just saying.
+from typing import ClassVar from unittest import mock from rest_framework.test import APITestCase from care.users.models import User from care.utils.assetintegration.usage_manager import UsageManager from care.utils.tests.test_utils import TestUtils -class UsageManagerTestCase(TestUtils, APITestCase): +class UsageManagerTestCase(TestUtils, APITestCase): + state: ClassVar + district: ClassVar + local_body: ClassVar + super_user: ClassVar + facility: ClassVar + asset_location: ClassVar + asset: ClassVar
11-20: I'm sure there's a perfectly good reason why this setup isn't documented.Adding a docstring would help others understand the purpose and structure of the test data setup. You know, for those rare occasions when someone else needs to maintain this code.
@classmethod def setUpTestData(cls): + """ + Sets up test data shared across all test methods. + Creates a hierarchical structure of: + State -> District -> Local Body -> Facility -> Asset Location -> Asset + Also creates a super user associated with the district. + """ cls.state = cls.create_state() cls.district = cls.create_district(cls.state) cls.local_body = cls.create_local_body(cls.district)
21-51: Magic numbers are like surprises - not everyone enjoys them.The user type assignment could be more explicit. Also, the cache mock setup could use some error handling, but I suppose that's not important enough to worry about.
+ USER_TYPE = User.TYPE_VALUE_MAP["StateAdmin"] + def setUp(self): self.user1 = self.create_user( username="test_user_1", state=self.state, district=self.district, - user_type=User.TYPE_VALUE_MAP["StateAdmin"], + user_type=self.USER_TYPE, )Also, consider adding error handling for cache mock setup:
def setUp(self): + try: - self.mock_cache = mock.MagicMock() - self.cache_patcher = mock.patch( - "care.utils.assetintegration.usage_manager.cache", self.mock_cache - ) - self.cache_patcher.start() + self.mock_cache = mock.MagicMock() + self.cache_patcher = mock.patch( + "care.utils.assetintegration.usage_manager.cache", self.mock_cache + ) + self.cache_patcher.start() + except Exception as e: + raise RuntimeError(f"Failed to setup cache mock: {e}")
91-102: A wild magic number appears in the timeout value.The 5-minute timeout (60 * 5) should probably be a constant. Also, it might be nice to test what happens when the timeout expires, but I'm sure you were getting to that.
+ CAMERA_LOCK_TIMEOUT = 60 * 5 # 5 minutes + def test_lock_camera(self): self.mock_cache.get.return_value = None self.assertTrue(self.usage_manager_user1.lock_camera()) self.mock_cache.set.assert_called_once_with( self.usage_manager_user1.current_user_cache_key, self.user1.id, - timeout=60 * 5, + timeout=self.CAMERA_LOCK_TIMEOUT, )care/facility/tests/test_consultation_bed_api.py (2)
432-446: Consider adding test cases for invalid privacy valuesWhile you've covered the empty request body case, it might be nice to also test how the endpoint handles invalid boolean values (like strings or numbers) in the
is_privacy_enabledfield.Add these test cases before the allowed users section:
response = self.client.patch( f"/api/v1/consultationbed/{consultation_bed.external_id}/set_privacy/", {"is_privacy_enabled": "not_a_boolean"}, ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) response = self.client.patch( f"/api/v1/consultationbed/{consultation_bed.external_id}/set_privacy/", {"is_privacy_enabled": 1}, ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
481-504: Add test for unauthenticated accessWhile you've thoroughly tested authenticated users with different permissions, it would be really great if you could also verify the behavior for unauthenticated requests.
Add this test case at the end:
# Clear any authenticated user self.client.force_authenticate(user=None) response = self.client.patch( f"/api/v1/consultationbed/{consultation_bed.external_id}/set_privacy/", {"is_privacy_enabled": True}, ) self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)care/utils/assetintegration/usage_manager.py (2)
54-54: Correct the typo in method name.The method
notify_waiting_list_on_asset_availabehas a typo in its name. It should benotify_waiting_list_on_asset_availablefor clarity and consistency.Apply this diff to fix the typo:
-def notify_waiting_list_on_asset_availabe(self) -> None: +def notify_waiting_list_on_asset_available(self) -> None:
121-122: Implement or remove thetake_controlmethod.The method
take_controlis currently empty. If it's not needed, consider removing it to keep the codebase clean.Would you like help in implementing this method, or should we remove it altogether?
care/utils/assetintegration/onvif.py (2)
59-61: Correct the typo in your error messages.The error message should read "Camera is currently in use by another user..." instead of "in used."
Apply this diff to fix the typo:
- "message": "Camera is currently in used by another user, you have been added to the waiting list for camera controls access", + "message": "Camera is currently in use by another user, you have been added to the waiting list for camera controls access",Also applies to: 97-99
71-73: Fix the duplicated word in the success message.The message contains a duplication: "Access to camera camera controls granted." Please correct it.
Here's the fix:
- "message": "Access to camera camera controls granted", + "message": "Access to camera controls granted",
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
📒 Files selected for processing (11)
care/facility/api/viewsets/asset.py(2 hunks)care/facility/api/viewsets/bed.py(2 hunks)care/facility/migrations/0467_consultationbed_is_privacy_enabled.py(1 hunks)care/facility/models/bed.py(1 hunks)care/facility/tests/test_asset_usage_manager.py(1 hunks)care/facility/tests/test_consultation_bed_api.py(2 hunks)care/utils/assetintegration/base.py(1 hunks)care/utils/assetintegration/hl7monitor.py(1 hunks)care/utils/assetintegration/onvif.py(4 hunks)care/utils/assetintegration/usage_manager.py(1 hunks)care/utils/assetintegration/ventilator.py(1 hunks)
🧰 Additional context used
🪛 Ruff (0.7.0)
care/facility/migrations/0467_consultationbed_is_privacy_enabled.py
9-9: Single quotes found but double quotes preferred
Replace single quotes with double quotes
(Q000)
9-9: Single quotes found but double quotes preferred
Replace single quotes with double quotes
(Q000)
14-14: Single quotes found but double quotes preferred
Replace single quotes with double quotes
(Q000)
15-15: Single quotes found but double quotes preferred
Replace single quotes with double quotes
(Q000)
🔇 Additional comments (10)
care/utils/assetintegration/hl7monitor.py (1)
23-23:
I see you added a user parameter... but forgot to use it?
The method signature has been updated to include the user parameter, but it's not being utilized in the implementation. This seems a bit... incomplete.
Let's check if other asset classes are using the user parameter:
care/utils/assetintegration/base.py (1)
22-23: Verify child class implementations
The signature change in this base class requires updates to all inheriting classes.
care/facility/models/bed.py (1)
95-97: Just a friendly reminder about those asset relationships...
The is_privacy_enabled field will likely affect asset access control, particularly with camera feeds. Since this PR is waiting for the camera plug development, we should verify the integration points.
Let's check for potential asset access patterns:
care/facility/tests/test_consultation_bed_api.py (1)
432-504: LGTM! The test coverage is comprehensive
The test implementation is well-structured and covers the essential scenarios for the privacy feature. The suggested improvements would make it even better, but the current implementation is already quite solid.
care/utils/assetintegration/usage_manager.py (1)
105-113: Double-check the logic in unlock_camera.
The method removes the current user from the cache and then from the waiting list, regardless of whether they held the lock. This might not be the intended behavior.
Please confirm if this logic is correct. If the current user isn't the one who locked the camera, we might not want to modify the cache.
care/utils/assetintegration/onvif.py (1)
36-36: Verify all calls to handle_action include the new user parameter.
With the addition of the user parameter to handle_action, ensure that all invocations of this method across the codebase are updated accordingly to prevent any unexpected errors.
Run the following script to find all usages of handle_action and check for the user parameter:
✅ Verification successful
All calls to handle_action are properly updated with the user parameter
The search results show that all implementations and the single invocation of handle_action are correctly using the user parameter:
- Implementations in:
care/utils/assetintegration/hl7monitor.pycare/utils/assetintegration/ventilator.pycare/utils/assetintegration/onvif.pycare/utils/assetintegration/base.py(base class)
- The only invocation found is in
care/facility/api/viewsets/asset.pywhich correctly passesrequest.user
I suppose we can all breathe a little easier knowing the codebase is perfectly consistent here.
🏁 Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash
# Description: Find all calls to `handle_action` and verify the inclusion of the `user` argument.
# Expected: All calls to `handle_action` should include the `user` parameter.
rg -A 2 'handle_action\(' care/
Length of output: 1130
care/facility/api/viewsets/bed.py (2)
8-8: Necessary import added
The action decorator is properly imported to support the new set_privacy method.
247-253: Reconsider the user permission logic
The use of or in the permission check is intriguing. This means that if either the user lacks update permission on the patient or their user type is in the specified list, they will be denied permission. It might be worthwhile to verify whether this aligns with the intended access control—should DistrictLabAdmin or StateLabAdmin users be denied even if they have update permission?
care/facility/api/viewsets/asset.py (2)
23-23: Importing PermissionDenied exception is appropriate
Including PermissionDenied from rest_framework.exceptions is necessary for handling permission-related errors in the viewset.
408-408: Passing request.user to handle_action enhances user context
By passing request.user to the handle_action method, you're ensuring that user-specific operations are correctly handled within the asset classes.
| @@ -20,7 +20,7 @@ def __init__(self, meta): | |||
| {key: f"{key} not found in asset metadata" for key in e.args} | |||
| ) from e | |||
|
|
|||
| def handle_action(self, action): | |||
| def handle_action(self, action, user): | |||
There was a problem hiding this comment.
💡 Codebase verification
Oh, I see what's happening with that user parameter now...
The user parameter is actually being used meaningfully in onvif.py with a UsageManager, while our other implementations in ventilator.py and hl7monitor.py are just... collecting dust with it. The base class defines it as part of the interface, so we should probably do something useful with it, don't you think?
care/utils/assetintegration/ventilator.py: User parameter is unusedcare/utils/assetintegration/hl7monitor.py: Same issue, parameter just sitting therecare/utils/assetintegration/onvif.py: Shows proper usage withUsageManager
Perhaps we could make these implementations a tad more security-conscious, like their ONVIF cousin?
🔗 Analysis chain
That's a lovely new user parameter you've added there... but are we going to use it?
The method signature has been updated to include the user parameter, but it's not being utilized in the implementation. This seems... incomplete.
Let's check if other asset classes are actually using this parameter:
🏁 Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash
# Search for handle_action implementations that use the user parameter
ast-grep --pattern 'def handle_action($_, $user) {
$$$
$user
$$$
}'
Length of output: 75
Script:
#!/bin/bash
# Let's try a broader search for handle_action implementations
rg "def handle_action" -A 5
# Also search for the base class implementation
rg "class BaseAssetIntegration" -A 10
Length of output: 2789
| @@ -92,6 +92,7 @@ class ConsultationBed(BaseModel): | |||
| bed = models.ForeignKey(Bed, on_delete=models.PROTECT, null=False, blank=False) | |||
| start_date = models.DateTimeField(null=False, blank=False) | |||
| end_date = models.DateTimeField(null=True, blank=True, default=None) | |||
| is_privacy_enabled = models.BooleanField(default=False) | |||
There was a problem hiding this comment.
Interesting choice to skip the validation rules...
The is_privacy_enabled field might benefit from some validation rules and state change tracking. Also, I couldn't help but notice the complete absence of test cases mentioned in the PR comments.
Consider adding:
- State change validation
- Audit logging for privacy changes
- Test cases covering privacy scenarios
class ConsultationBed(BaseModel):
+ def clean(self):
+ super().clean()
+ if self.is_privacy_enabled and not self.consultation.facility.privacy_enabled:
+ raise ValidationError(
+ {"is_privacy_enabled": "Facility must have privacy features enabled"}
+ )
+
+ def save(self, *args, **kwargs):
+ self.full_clean()
+ if self.tracker.has_changed('is_privacy_enabled'):
+ # Log privacy state change
+ pass
+ super().save(*args, **kwargs)Committable suggestion skipped: line range outside the PR's diff.
| def test_has_access(self): | ||
| self.mock_cache.get.return_value = None | ||
| self.assertTrue(self.usage_manager_user1.has_access()) | ||
|
|
||
| self.mock_cache.get.return_value = self.user1.id | ||
| self.assertTrue(self.usage_manager_user1.has_access()) | ||
|
|
||
| self.mock_cache.get.return_value = self.user2.id | ||
| self.assertFalse(self.usage_manager_user1.has_access()) | ||
|
|
||
| def test_unlock_camera(self): | ||
| self.mock_cache.get.return_value = self.user1.id | ||
|
|
||
| with mock.patch.object( | ||
| self.usage_manager_user1, "notify_waiting_list_on_asset_availabe" | ||
| ) as mock_notify: | ||
| self.usage_manager_user1.unlock_camera() | ||
|
|
||
| self.mock_cache.delete.assert_called_once_with( | ||
| self.usage_manager_user1.current_user_cache_key | ||
| ) | ||
|
|
||
| mock_notify.assert_called_once() | ||
|
|
||
| def test_request_access(self): | ||
| self.mock_cache.get.return_value = None | ||
| self.assertTrue(self.usage_manager_user1.request_access()) | ||
|
|
||
| self.mock_cache.get.return_value = self.user2.id | ||
| with mock.patch( | ||
| "care.utils.notification_handler.send_webpush" | ||
| ) as mock_send_webpush: | ||
| result = self.usage_manager_user1.request_access() | ||
| self.assertFalse(result) | ||
| mock_send_webpush.assert_called_once() | ||
|
|
||
| def test_lock_camera(self): | ||
| self.mock_cache.get.return_value = None | ||
| self.assertTrue(self.usage_manager_user1.lock_camera()) | ||
| self.mock_cache.set.assert_called_once_with( | ||
| self.usage_manager_user1.current_user_cache_key, | ||
| self.user1.id, | ||
| timeout=60 * 5, | ||
| ) | ||
|
|
||
| self.mock_cache.get.return_value = self.user2.id | ||
| self.assertFalse(self.usage_manager_user1.lock_camera()) | ||
|
|
||
| def test_current_user(self): | ||
| self.mock_cache.get.return_value = self.user1.id | ||
|
|
||
| mock_serializer = mock.MagicMock() | ||
| mock_serializer.data = { | ||
| "id": self.user1.id, | ||
| "username": self.user1.username, | ||
| } | ||
|
|
||
| with mock.patch( | ||
| "care.facility.api.serializers.asset.UserBaseMinimumSerializer", | ||
| return_value=mock_serializer, | ||
| ): | ||
| current_user_data = self.usage_manager_user1.current_user() | ||
| self.assertIsNotNone(current_user_data) | ||
| self.assertEqual(current_user_data["id"], self.user1.id) |
There was a problem hiding this comment.
🛠️ Refactor suggestion
The happy path tests look great! Though I can't help but notice some missing scenarios.
Consider adding tests for:
- Redis connection failures
- Concurrent access attempts
- Cache timeout expiration
- Invalid user states
I'm sure these edge cases aren't important though... until they are.
Here's a sample test case for Redis failure:
def test_lock_camera_redis_failure(self):
self.mock_redis_client.ping.side_effect = ConnectionError("Redis connection failed")
with self.assertRaises(ConnectionError):
self.usage_manager_user1.lock_camera()| cls.asset = cls.create_asset(cls.asset_location) | ||
|
|
||
| def setUp(self): | ||
| self.user1 = self.create_user( | ||
| username="test_user_1", | ||
| state=self.state, | ||
| district=self.district, | ||
| user_type=User.TYPE_VALUE_MAP["StateAdmin"], | ||
| ) | ||
| self.user2 = self.create_user( | ||
| username="test_user_2", | ||
| state=self.state, | ||
| district=self.district, | ||
| user_type=User.TYPE_VALUE_MAP["StateAdmin"], | ||
| ) | ||
|
|
||
| self.mock_cache = mock.MagicMock() | ||
| self.cache_patcher = mock.patch( | ||
| "care.utils.assetintegration.usage_manager.cache", self.mock_cache | ||
| ) | ||
| self.cache_patcher.start() | ||
|
|
||
| self.usage_manager_user1 = UsageManager( | ||
| asset_id=self.asset.external_id, user=self.user1 | ||
| ) | ||
| self.usage_manager_user2 = UsageManager( | ||
| asset_id=self.asset.external_id, user=self.user2 | ||
| ) | ||
|
|
||
| self.mock_redis_client = mock.MagicMock() | ||
| self.usage_manager_user1.redis_client = self.mock_redis_client | ||
| self.usage_manager_user2.redis_client = self.mock_redis_client | ||
|
|
||
| def tearDown(self): | ||
| self.cache_patcher.stop() | ||
|
|
||
| def test_has_access(self): | ||
| self.mock_cache.get.return_value = None | ||
| self.assertTrue(self.usage_manager_user1.has_access()) | ||
|
|
||
| self.mock_cache.get.return_value = self.user1.id | ||
| self.assertTrue(self.usage_manager_user1.has_access()) | ||
|
|
||
| self.mock_cache.get.return_value = self.user2.id | ||
| self.assertFalse(self.usage_manager_user1.has_access()) | ||
|
|
||
| def test_unlock_camera(self): | ||
| self.mock_cache.get.return_value = self.user1.id | ||
|
|
||
| with mock.patch.object( | ||
| self.usage_manager_user1, "notify_waiting_list_on_asset_availabe" | ||
| ) as mock_notify: | ||
| self.usage_manager_user1.unlock_camera() | ||
|
|
||
| self.mock_cache.delete.assert_called_once_with( | ||
| self.usage_manager_user1.current_user_cache_key | ||
| ) | ||
|
|
||
| mock_notify.assert_called_once() | ||
|
|
||
| def test_request_access(self): | ||
| self.mock_cache.get.return_value = None | ||
| self.assertTrue(self.usage_manager_user1.request_access()) | ||
|
|
||
| self.mock_cache.get.return_value = self.user2.id | ||
| with mock.patch( | ||
| "care.utils.notification_handler.send_webpush" | ||
| ) as mock_send_webpush: | ||
| result = self.usage_manager_user1.request_access() | ||
| self.assertFalse(result) | ||
| mock_send_webpush.assert_called_once() | ||
|
|
||
| def test_lock_camera(self): | ||
| self.mock_cache.get.return_value = None | ||
| self.assertTrue(self.usage_manager_user1.lock_camera()) | ||
| self.mock_cache.set.assert_called_once_with( | ||
| self.usage_manager_user1.current_user_cache_key, | ||
| self.user1.id, | ||
| timeout=60 * 5, | ||
| ) | ||
|
|
||
| self.mock_cache.get.return_value = self.user2.id | ||
| self.assertFalse(self.usage_manager_user1.lock_camera()) | ||
|
|
||
| def test_current_user(self): | ||
| self.mock_cache.get.return_value = self.user1.id | ||
|
|
||
| mock_serializer = mock.MagicMock() | ||
| mock_serializer.data = { | ||
| "id": self.user1.id, | ||
| "username": self.user1.username, | ||
| } | ||
|
|
||
| with mock.patch( | ||
| "care.facility.api.serializers.asset.UserBaseMinimumSerializer", | ||
| return_value=mock_serializer, | ||
| ): | ||
| current_user_data = self.usage_manager_user1.current_user() | ||
| self.assertIsNotNone(current_user_data) | ||
| self.assertEqual(current_user_data["id"], self.user1.id) |
There was a problem hiding this comment.
💡 Codebase verification
Several methods seem to have slipped through the testing net
While the existing tests are... fine, there appear to be some methods that haven't received the attention they deserve:
get_waiting_listadd_to_waiting_listremove_from_waiting_listclear_waiting_listnotify_current_user_on_request_accesstake_control
I'm sure there's a perfectly good reason these methods don't have corresponding test cases, but it would be absolutely wonderful if we could ensure they work as intended. After all, untested code is like a mystery box - you never know what surprises it might contain.
🔗 Analysis chain
The test suite looks... adequate.
While the basic functionality is covered, it would be really nice if we could verify the test coverage percentage. You know, just to make sure we're not missing anything important.
🏁 Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash
# Check test coverage for the UsageManager class
# Find the corresponding implementation file
implementation_file=$(fd -t f -e py usage_manager)
# Run coverage report if available
if command -v coverage &> /dev/null; then
coverage run -m pytest "$implementation_file"
coverage report --include="$implementation_file"
else
echo "Coverage tool not found. But I'm sure the coverage is fine... probably."
fi
Length of output: 325
Script:
#!/bin/bash
# Let's examine the test coverage by analyzing the implementation and test files
# First, let's see what methods are defined in the implementation
echo "=== Implementation Methods ==="
ast-grep --pattern 'def $_($$$) {$$$}' care/utils/assetintegration/usage_manager.py
echo -e "\n=== Test Methods ==="
ast-grep --pattern 'def test_$_($$$) {$$$}' care/facility/tests/test_asset_usage_manager.py
# Let's also check for any untested public methods
echo -e "\n=== Public Methods in Implementation ==="
rg "^[[:space:]]*def [^_].*\(" care/utils/assetintegration/usage_manager.py
Length of output: 990
| allowed_user_types = [ | ||
| "StateAdmin", | ||
| "DistrictAdmin", | ||
| "Doctor", | ||
| ] | ||
| for user_type in allowed_user_types: | ||
| user = self.create_user( | ||
| username="test_user_" + user_type.lower(), | ||
| state=self.state, | ||
| district=self.district, | ||
| user_type=User.TYPE_VALUE_MAP[user_type], | ||
| home_facility=self.facility, | ||
| ) | ||
|
|
||
| self.client.force_authenticate(user) | ||
|
|
||
| response = self.client.patch( | ||
| f"/api/v1/consultationbed/{consultation_bed.external_id}/set_privacy/", | ||
| {"is_privacy_enabled": True}, | ||
| ) | ||
| self.assertEqual(response.status_code, status.HTTP_200_OK) | ||
| self.assertTrue( | ||
| ConsultationBed.objects.get(id=consultation_bed.id).is_privacy_enabled | ||
| ) | ||
|
|
||
| response = self.client.patch( | ||
| f"/api/v1/consultationbed/{consultation_bed.external_id}/set_privacy/", | ||
| {"is_privacy_enabled": False}, | ||
| ) | ||
| self.assertEqual(response.status_code, status.HTTP_200_OK) | ||
| self.assertFalse( | ||
| ConsultationBed.objects.get(id=consultation_bed.id).is_privacy_enabled | ||
| ) | ||
|
|
There was a problem hiding this comment.
🛠️ Refactor suggestion
Add test for concurrent privacy modifications
The current tests are almost perfect, but they don't verify how the endpoint handles concurrent modifications. This could lead to race conditions in production.
Add this test case to check for race conditions:
def test_set_privacy_concurrent_modifications(self):
consultation_bed = ConsultationBed.objects.create(
consultation=self.consultation,
bed=self.bed1,
start_date=timezone.now(),
is_privacy_enabled=False,
)
# Simulate concurrent modifications
bed1 = ConsultationBed.objects.get(id=consultation_bed.id)
bed2 = ConsultationBed.objects.get(id=consultation_bed.id)
bed1.is_privacy_enabled = True
bed1.save()
bed2.is_privacy_enabled = False
bed2.save()
# Verify final state
final_state = ConsultationBed.objects.get(id=consultation_bed.id)
self.assertFalse(final_state.is_privacy_enabled)| { | ||
| "type": "MESSAGE", | ||
| "asset_id": self.asset, | ||
| "message": f"{User.REVERSE_TYPE_MAP[requester.user_type]}, {requester.full_name} ({requester.username}) has requested access to the camera", |
There was a problem hiding this comment.
Sanitize user input in notification messages.
Directly inserting user-provided data into messages can lead to security vulnerabilities. It's safer to sanitize or escape user inputs.
Consider updating the message construction:
"message": f"{User.REVERSE_TYPE_MAP.get(requester.user_type, 'User')}, {requester.full_name} ({requester.username}) has requested access to the camera",Ensure all user attributes are properly sanitized.
Committable suggestion skipped: line range outside the PR's diff.
| LOCK_CAMERA = "lock_camera" | ||
| UNLOCK_CAMERA = "unlock_camera" | ||
| REQUEST_ACCESS = "request_access" | ||
| TAKE_CONTROL = "take_control" | ||
|
|
There was a problem hiding this comment.
Implement handling for the TAKE_CONTROL action.
The TAKE_CONTROL action is defined in the OnvifActions enum but lacks corresponding handling in the handle_action method. This could lead to a ValidationError if the action is invoked. It might be helpful to implement this action or remove it if it's not needed.
You can apply this diff to remove the unused action:
- TAKE_CONTROL = "take_control"Alternatively, if you plan to implement it later, consider adding a # TODO comment as a reminder.
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| LOCK_CAMERA = "lock_camera" | |
| UNLOCK_CAMERA = "unlock_camera" | |
| REQUEST_ACCESS = "request_access" | |
| TAKE_CONTROL = "take_control" | |
| LOCK_CAMERA = "lock_camera" | |
| UNLOCK_CAMERA = "unlock_camera" | |
| REQUEST_ACCESS = "request_access" | |
| if action_type == self.OnvifActions.GET_STREAM_TOKEN.value: | ||
| return self.api_post( | ||
| self.get_url("api/stream/getToken/videoFeed"), | ||
| { | ||
| "stream_id": self.access_key, | ||
| }, | ||
| ) |
There was a problem hiding this comment.
Ensure access control for the GET_STREAM_TOKEN action.
The GET_STREAM_TOKEN action does not currently check for user access before returning the stream token. This could allow unauthorized users to obtain the stream token.
Consider adding an access check before handling this action:
if action_type == self.OnvifActions.GET_STREAM_TOKEN.value:
+ if not camera_manager.has_access():
+ raise PermissionDenied(
+ {
+ "message": "Camera is currently in use by another user, you have been added to the waiting list for camera controls access",
+ "camera_user": camera_manager.current_user(),
+ }
+ )
return self.api_post(
self.get_url("api/stream/getToken/videoFeed"),
{Committable suggestion skipped: line range outside the PR's diff.
| if is_privacy_enabled is None and not isinstance(is_privacy_enabled, bool): | ||
| return Response( | ||
| {"detail": "is_privacy_enabled is required and should be a boolean"}, | ||
| status=status.HTTP_400_BAD_REQUEST, | ||
| ) |
There was a problem hiding this comment.
Improve the validation condition for is_privacy_enabled
Interesting approach on the condition here. However, it might not catch all invalid inputs. It would be better to check whether is_privacy_enabled is None or not a boolean value. Currently, the condition only triggers if is_privacy_enabled is both None and not a boolean, which is unnecessary since None isn't a boolean anyway.
Apply this diff to correct the condition:
- if is_privacy_enabled is None and not isinstance(is_privacy_enabled, bool):
+ if is_privacy_enabled is None or not isinstance(is_privacy_enabled, bool):📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| if is_privacy_enabled is None and not isinstance(is_privacy_enabled, bool): | |
| return Response( | |
| {"detail": "is_privacy_enabled is required and should be a boolean"}, | |
| status=status.HTTP_400_BAD_REQUEST, | |
| ) | |
| if is_privacy_enabled is None or not isinstance(is_privacy_enabled, bool): | |
| return Response( | |
| {"detail": "is_privacy_enabled is required and should be a boolean"}, | |
| status=status.HTTP_400_BAD_REQUEST, | |
| ) |
| except PermissionDenied as e: | ||
| return Response( | ||
| {**e.detail}, | ||
| status=status.HTTP_403_FORBIDDEN, | ||
| ) | ||
|
|
There was a problem hiding this comment.
Potential issue with unpacking e.detail
Unpacking e.detail using {**e.detail} assumes it's a dictionary, but e.detail might be a string. This could raise a TypeError. You might want to verify the type of e.detail and adjust accordingly to ensure proper error handling.
Apply this diff to fix the issue:
except PermissionDenied as e:
- return Response(
- {**e.detail},
- status=status.HTTP_403_FORBIDDEN,
- )
+ return Response(
+ {'detail': str(e.detail)},
+ status=status.HTTP_403_FORBIDDEN,
+ )📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| except PermissionDenied as e: | |
| return Response( | |
| {**e.detail}, | |
| status=status.HTTP_403_FORBIDDEN, | |
| ) | |
| except PermissionDenied as e: | |
| return Response( | |
| {'detail': str(e.detail)}, | |
| status=status.HTTP_403_FORBIDDEN, | |
| ) |
Associated Issue
@ohcnetwork/care-backend-maintainers @ohcnetwork/care-backend-admins
Summary by CodeRabbit
Release Notes
New Features
Bug Fixes
Documentation
Chores