feat: allow reads to be filtered by originating plugin

synse_server/api/http.py

@@ -6,7 +6,7 @@
 from sanic.response import HTTPResponse, StreamingHTTPResponse, stream
 from structlog import get_logger
 
-from synse_server import cmd, errors, utils
+from synse_server import cmd, errors, plugin, utils
 
 logger = get_logger()
 
@@ -99,7 +99,7 @@ async def plugins(request: Request) -> HTTPResponse:
 
 
 @v3.route('/plugin/<plugin_id>')
-async def plugin(request: Request, plugin_id: str) -> HTTPResponse:
+async def plugin_info(request: Request, plugin_id: str) -> HTTPResponse:
     """Get detailed information on the specified plugin.
 
     URI Parameters:
@@ -282,6 +282,8 @@ async def read(request: Request) -> HTTPResponse:
             group only selects devices which match all of the tags in the group. If
             multiple tag groups are specified, the result is the union of the matches
             from each individual tag group.
+        plugin: The ID of the plugin to get device readings from. If not specified,
+            all plugins are considered valid for reading.
 
     HTTP Codes:
         * 200: OK
@@ -303,11 +305,18 @@ async def read(request: Request) -> HTTPResponse:
         for group in param_tags:
             tag_groups.append(group.split(','))
 
+    plugin_id = request.args.get('plugin', None)
+    if plugin_id and plugin_id not in plugin.manager.plugins:
+        raise errors.InvalidUsage(
+            'invalid parameter: specified plugin ID does not correspond with known plugin',
+        )
+
     try:
         return utils.http_json_response(
             await cmd.read(
                 ns=namespace,
                 tag_groups=tag_groups,
+                plugin_id=plugin_id,
             ),
         )
     except Exception:

synse_server/cmd/read.py

@@ -2,7 +2,7 @@
 import asyncio
 import queue
 import threading
-from typing import Any, AsyncIterable, Dict, List, Union
+from typing import Any, AsyncIterable, Dict, List, Optional, Union
 
 import synse_grpc.utils
 import websockets
@@ -48,7 +48,11 @@ def reading_to_dict(reading: api.V3Reading) -> Dict[str, Any]:
     }
 
 
-async def read(ns: str, tag_groups: Union[List[str], List[List[str]]]) -> List[Dict[str, Any]]:
+async def read(
+        ns: str,
+        tag_groups: Union[List[str], List[List[str]]],
+        plugin_id: Optional[str] = None,
+) -> List[Dict[str, Any]]:
     """Generate the readings response data.
 
     Args:
@@ -57,6 +61,8 @@ async def read(ns: str, tag_groups: Union[List[str], List[List[str]]]) -> List[D
             is ignored.
         tag_groups: The tags groups used to filter devices. If no tag
             groups are given (and thus no tags), no filtering is done.
+        plugin_id: The ID of the plugin to get device readings from. If not specified,
+            all plugins are considered valid for reading.
 
     Returns:
         A list of dictionary representations of device reading response(s).
@@ -68,6 +74,14 @@ async def read(ns: str, tag_groups: Union[List[str], List[List[str]]]) -> List[D
         logger.debug('no tags specified, reading with no tag filter', command='READ')
         readings = []
         for p in plugin.manager:
+            if plugin_id and p.id != plugin_id:
+                logger.debug(
+                    'skipping plugin for read - plugin filter set',
+                    filter=plugin_id,
+                    skipped=p.id,
+                )
+                continue
+
             if not p.active:
                 logger.debug(
                     'plugin not active, will not read its devices',
@@ -104,6 +118,14 @@ async def read(ns: str, tag_groups: Union[List[str], List[List[str]]]) -> List[D
                 group[i] = f'{ns}/{tag}'
 
         for p in plugin.manager:
+            if plugin_id and p.id != plugin_id:
+                logger.debug(
+                    'skipping plugin for read - plugin filter set',
+                    filter=plugin_id,
+                    skipped=p.id,
+                )
+                continue
+
             if not p.active:
                 logger.debug(
                     'plugin not active, will not read its devices',

tests/unit/api/test_http.py

@@ -898,6 +898,7 @@ def test_ok(self, synse_app):
         mock_cmd.assert_called_with(
             ns='default',
             tag_groups=[],
+            plugin_id=None,
         )
 
     def test_error(self, synse_app):
@@ -920,6 +921,7 @@ def test_error(self, synse_app):
         mock_cmd.assert_called_with(
             ns='default',
             tag_groups=[],
+            plugin_id=None,
         )
 
     def test_invalid_multiple_ns(self, synse_app):
@@ -974,6 +976,7 @@ def test_param_tags(self, synse_app, qparam, expected):
         mock_cmd.assert_called_with(
             ns='default',
             tag_groups=expected,
+            plugin_id=None
         )
 
     @pytest.mark.parametrize(
@@ -1009,8 +1012,45 @@ def test_param_ns(self, synse_app, qparam, expected):
         mock_cmd.assert_called_with(
             ns=expected,
             tag_groups=[],
+            plugin_id=None,
         )
 
+    def test_param_plugin(self, synse_app, mocker):
+        with asynctest.patch('synse_server.cmd.read') as mock_cmd:
+            mock_cmd.return_value = [{'value': 1, 'type': 'temperature'}]
+            mocker.patch.dict('synse_server.plugin.manager.plugins', {
+                '123456': None,
+            })
+
+            resp = synse_app.test_client.get(
+                '/v3/read?plugin=123456',
+                gather_request=False,
+            )
+            assert resp.status == 200
+            assert resp.headers['Content-Type'] == 'application/json'
+
+            body = ujson.loads(resp.body)
+            assert body == mock_cmd.return_value
+
+        mock_cmd.assert_called_once_with(
+            ns='default',
+            tag_groups=[],
+            plugin_id='123456',
+        )
+
+    def test_param_plugin_no_plugin(self, synse_app):
+        with asynctest.patch('synse_server.cmd.read') as mock_cmd:
+            mock_cmd.return_value = [{'value': 1, 'type': 'temperature'}]
+
+            resp = synse_app.test_client.get(
+                '/v3/read?plugin=123456',
+                gather_request=False,
+            )
+            assert resp.status == 400
+            assert resp.headers['Content-Type'] == 'application/json'
+
+        mock_cmd.assert_not_called()
+
 
 @pytest.mark.usefixtures('patch_utils_rfc3339now')
 class TestV3ReadCache:

tests/unit/cmd/test_read.py

@@ -314,6 +314,80 @@ async def test_read_ok_single_tag_group_without_ns(mocker, simple_plugin, state_
     ])
 
 
+@pytest.mark.asyncio
+async def test_read_ok_single_tag_group_without_ns_with_plugin(
+        mocker, simple_plugin, state_reading):
+
+    # Mock test data
+    mocker.patch.dict('synse_server.plugin.PluginManager.plugins', {
+        '123': simple_plugin,
+        '456': simple_plugin,
+    })
+
+    mock_read = mocker.patch(
+        'synse_grpc.client.PluginClientV3.read',
+        return_value=[
+            state_reading,
+        ],
+    )
+
+    # --- Test case -----------------------------
+    # Set the simple_plugin to active to start.
+    simple_plugin.active = True
+
+    resp = await cmd.read('default', ['foo', 'bar', 'vapor/ware'], plugin_id='123')
+
+    # Note: There are two plugins defined, but only one is targeted, so we should expect
+    # only one plugin to return a reading.
+    assert resp == [
+        {  # from state_reading fixture
+            'device': 'ccc',
+            'timestamp': '2019-04-22T13:30:00Z',
+            'type': 'state',
+            'device_type': 'led',
+            'value': 'on',
+            'unit': None,
+            'context': {},
+        },
+    ]
+
+    assert simple_plugin.active is True
+
+    mock_read.assert_called()
+    mock_read.assert_has_calls([
+        mocker.call(tags=['default/foo', 'default/bar', 'vapor/ware']),
+    ])
+
+
+@pytest.mark.asyncio
+async def test_read_ok_with_unknown_plugin(mocker, simple_plugin, state_reading):
+    # Mock test data
+    mocker.patch.dict('synse_server.plugin.PluginManager.plugins', {
+        '123': simple_plugin,
+        '456': simple_plugin,
+    })
+
+    mock_read = mocker.patch(
+        'synse_grpc.client.PluginClientV3.read',
+        return_value=[
+            state_reading,
+        ],
+    )
+
+    # --- Test case -----------------------------
+    # Set the simple_plugin to active to start.
+    simple_plugin.active = True
+
+    resp = await cmd.read('default', ['foo', 'bar', 'vapor/ware'], plugin_id='666')
+
+    # Note: No plugin matched id 666, so we should not expect to get any data back.
+    assert resp == []
+
+    assert simple_plugin.active is True
+
+    mock_read.assert_not_called()
+
+
 @pytest.mark.asyncio
 async def test_read_device_not_found():
     with asynctest.patch('synse_server.cache.get_plugin') as mock_get: