| # Copyright 2017 The ChromiumOS Authors |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| """Invoke remote procedure call for interaction with shopfloor backend. |
| |
| Description |
| ----------- |
| The Chromium OS Factory Software has defined a protocol, "Chrome OS Factory |
| Shopfloor Service Specification", to access factory manufacturing line shopfloor |
| system (or MES) backend system. This test allows interaction with a shopfloor |
| service, by requesting the factory server with defined protocol. |
| |
| For more information about Chrome OS Factory Shopfloor Service Specification, |
| read |
| https://chromium.googlesource.com/chromiumos/platform/factory/+/HEAD/py/shopfloor/README.md |
| |
| By default, the protocol has been simplified so you don't need to manually |
| generate or process ``FactoryDeviceData`` or ``DeviceData`` - just provide |
| the constant arguments from test list. |
| |
| For example, the method ``NotifyStart(data, station)`` can be invoked by |
| (assume station is ``'SMT'``) ``method='NotifyStart'`` and ``args=['SMT']``. |
| Also the return value is automatically merged into Device Data (see |
| ``cros.factory.test.device_data`` for more details). |
| |
| For OEM Chromebook projects, you should only use the standard methods defined in |
| Chrome OS Factory Shopfloor Service Specification. However, if you need to work |
| on a customized project or using a fixture with XMLRPC interface, it is possible |
| to use this test by setting argument ``raw_invocation`` to True. |
| |
| When ``raw_invocation`` is True, the invocation will simply run with argument |
| ``args`` and ``kargs``, no auto-generation of FactoryDeviceData or DeviceData. |
| The return value will still be merged to device data. |
| |
| Test Procedure |
| -------------- |
| This is an automated test without user interaction unless manually 'retry' is |
| needed. |
| |
| When started, the test will connect to remote server and try to invoke specified |
| method with given arguments, and will display return (error) messages and wait |
| for retry on failure. |
| |
| The pytest also sets device data to indicate certain checkpoint is passed, such |
| as ``factory.start_SMT``, ``factory.end_SMT``. These values can be synced to RW |
| VPD by write_device_data_to_vpd.py. |
| |
| Dependency |
| ---------- |
| No special dependency on client side, but the server must be implemented with |
| needed XMLRPC methods. |
| |
| Examples |
| -------- |
| To start 'SMT' station tests, add this in test list:: |
| |
| { |
| "pytest_name": "shopfloor_service", |
| "args": { |
| "args": ["SMT"], |
| "method": "NotifyStart" |
| } |
| } |
| |
| To invoke a non-standard call 'DoSomething' with args (1, 2) and keyword args |
| {'arg1': 1}:: |
| |
| { |
| "pytest_name": "shopfloor_service", |
| "args": { |
| "args": [1, 2], |
| "raw_invocation": true, |
| "kargs": { |
| "arg1": 1 |
| }, |
| "method": "DoSomething" |
| } |
| } |
| |
| To manually set factory server URL, set 'server_url' in test list:: |
| |
| { |
| "pytest_name": "shopfloor_service", |
| "args": { |
| "args": ["arg1", "arg2"], |
| "method": "DoSomething", |
| "server_url": "http://my-server:8080" |
| } |
| } |
| |
| To auto-detect factory server by received DHCP IP address, specify a mapping |
| object with key set to "IP/CIDR" and value set to server URL:: |
| |
| { |
| "pytest_name": "shopfloor_service", |
| "args": { |
| "args": ["arg1", "arg2"], |
| "method": "DoSomething", |
| "server_url": { |
| "192.168.1.0/24": "http://192.168.1.254:8080", |
| "10.3.0.0/24": "http://10.3.0.10:8080", |
| "10.1.0.0/16": "http://10.1.0.10:8080", |
| "default": "http://my-default-factory-server:8080" |
| } |
| } |
| } |
| """ |
| |
| |
| import collections.abc |
| import logging |
| import pprint |
| import threading |
| |
| from cros.factory.device import device_utils |
| from cros.factory.test import device_data |
| from cros.factory.test.i18n import _ |
| from cros.factory.test.rules import privacy |
| from cros.factory.test import server_proxy |
| from cros.factory.test import test_case |
| from cros.factory.test import test_ui |
| from cros.factory.test.utils.url_spec import URLSpec |
| from cros.factory.utils.arg_utils import Arg |
| from cros.factory.utils import debug_utils |
| from cros.factory.utils import log_utils |
| from cros.factory.utils import process_utils |
| from cros.factory.utils import shelve_utils |
| from cros.factory.utils import webservice_utils |
| |
| |
| class ServiceSpec: |
| """The specification of shopfloor service API.""" |
| |
| def __init__(self, has_data=True, auto_values=None, data_args=None, |
| has_privacy_args=False): |
| self.has_data = has_data |
| self.has_privacy_args = has_privacy_args |
| self.auto_values = auto_values or {} |
| self.data_args = data_args |
| |
| |
| class ShopfloorService(test_case.TestCase): |
| """Execution of remote shoploor service.""" |
| related_components = tuple() |
| |
| |
| ARGS = [ |
| Arg('method', str, 'Name of shopfloor service method to call'), |
| Arg('args', list, 'Arguments for specified method.', default=None), |
| Arg('kargs', collections.abc.Mapping, 'Keyword arguments for method.', |
| default=None), |
| Arg('raw_invocation', bool, 'Allow invocation of arbitrary calls.', |
| default=False), |
| Arg('server_url', (str, dict), 'The URL to factory server', default=None), |
| ] |
| |
| # The expected value for GetVersion, to help checking server implementation. |
| SERVICE_VERSION = '1.0' |
| |
| # Domain of values to exchange. |
| DOMAIN_SERIALS = device_data.KEY_SERIALS |
| DOMAIN_FACTORY = device_data.KEY_FACTORY |
| DOMAIN_VPD = device_data.KEY_VPD |
| DOMAIN_COMPONENT = device_data.KEY_COMPONENT |
| KEY_HWID = device_data.KEY_HWID |
| KEY_FM = device_data.KEY_FM |
| |
| KEY_VPD_USER_ECHO = device_data.KEY_VPD_USER_REGCODE |
| KEY_VPD_GROUP_ECHO = device_data.KEY_VPD_GROUP_REGCODE |
| |
| # Service API method names defined in version 1.0, in {name: has_data} format. |
| METHODS = { |
| 'GetVersion': ServiceSpec(has_data=False), |
| 'NotifyStart': ServiceSpec(auto_values={'factory.start_{1}': True}), |
| 'NotifyEnd': ServiceSpec(auto_values={'factory.end_{1}': True}), |
| 'NotifyEvent': ServiceSpec(auto_values={'factory.event_{1}': True}), |
| 'GetDeviceInfo': ServiceSpec(), |
| 'ActivateRegCode': ServiceSpec( |
| has_data=False, has_privacy_args=True, |
| auto_values={'factory.activate_reg_code': True}, |
| data_args=[KEY_VPD_USER_ECHO, KEY_VPD_GROUP_ECHO, KEY_HWID]), |
| 'UpdateTestResult': ServiceSpec(), |
| } |
| |
| def setUp(self): |
| self.dut = device_utils.CreateDUTInterface() |
| self.event = threading.Event() |
| # yapf: disable |
| self.ui.ToggleTemplateClass('font-large', True) # type: ignore #TODO(b/338318729) Fixit! # pylint: disable=line-too-long |
| # yapf: enable |
| |
| def GetFactoryDeviceData(self): |
| """Returns a dictionary in FactoryDeviceData format.""" |
| data = {} |
| # Warning: DO NOT ADD ANY EXTRA DOMAINS HERE WITHOUT REVIEW. |
| # Any protocol here must be compliant to Chrome OS Factory Shopfloor Service |
| # Specification: |
| # https://chromium.googlesource.com/chromiumos/platform/factory/+/HEAD/py/shopfloor/README.md |
| # Extra fields may cause security or privacy concern, and fail other |
| # partners sharing same factory branch. Especially that DOMAIN_VPD cannot be |
| # added since that would break privacy concern by registration (ECHO) codes. |
| for domain in [self.DOMAIN_SERIALS, self.DOMAIN_FACTORY]: |
| flat_data = device_data.FlattenData( |
| device_data.GetDeviceData(domain, {}), domain) |
| data.update(flat_data) |
| hwid = device_data.GetDeviceData(self.KEY_HWID, |
| self.dut.CallOutput('crossystem hwid')) |
| if hwid: |
| data[self.KEY_HWID] = hwid |
| return data |
| |
| def UpdateAutoResults(self, method, result, args): |
| """Updates auto values (based on method) to results.""" |
| auto_values = self.METHODS[method].auto_values |
| for k, v in auto_values.items(): |
| result[k.format(*args)] = v |
| |
| def UpdateDeviceData(self, data): |
| """Updates system device data according to the given data.""" |
| prefixes = [ |
| self.DOMAIN_SERIALS, self.DOMAIN_VPD, self.DOMAIN_COMPONENT, |
| self.DOMAIN_FACTORY, self.KEY_HWID, self.KEY_FM |
| ] |
| illegal_keys = [k for k in data if k.partition('.')[0] not in prefixes] |
| if illegal_keys: |
| raise ValueError(f'Invalid response keys: {illegal_keys!r}') |
| keys_to_delete = [k for k, v in data.items() if v is None] |
| device_data.DeleteDeviceData(keys_to_delete) |
| data = {k: v for k, v in data.items() if k not in keys_to_delete} |
| device_data.UpdateDeviceData(data) |
| |
| @classmethod |
| def FilterDict(cls, data): |
| """Returns a dict with privacy data filtered.""" |
| result = shelve_utils.DictShelfView(shelve_utils.InMemoryShelf()) |
| for k, v in data.items(): |
| result.SetValue(k, v) |
| if not result.GetKeys(): |
| return {} |
| return privacy.FilterDict(result.GetValue('')) |
| |
| def ShowMessage(self, caption, css, message, retry=False): |
| retry_button = [ |
| '<button data-test-event="retry">', |
| _('Retry'), '</button>' |
| ] if retry else '' |
| # yapf: disable |
| self.ui.SetState([ # type: ignore #TODO(b/338318729) Fixit! # pylint: disable=line-too-long |
| # yapf: enable |
| f'<span class="{css}">', caption, |
| '</span><p><textarea rows=25 cols=90 readonly>', |
| test_ui.Escape(message, False), '</textarea><p>', retry_button |
| ]) |
| |
| def HandleError(self, message, invocation_message): |
| self.ShowMessage( |
| _('Shopfloor exception:'), 'test-status-failed large', '\n'.join( |
| (message.splitlines()[-1], invocation_message, message)), True) |
| process_utils.WaitEvent(self.event) |
| self.event.clear() |
| |
| def runTest(self): |
| # yapf: disable |
| self.event_loop.AddEventHandler( # type: ignore #TODO(b/338318729) Fixit! # pylint: disable=line-too-long |
| # yapf: enable |
| 'retry', lambda unused_event: self.event.set()) |
| # yapf: disable |
| server_url = URLSpec.FindServerURL(self.args.server_url, self.dut) # type: ignore #TODO(b/338318729) Fixit! # pylint: disable=line-too-long |
| # yapf: enable |
| if server_url: |
| server = webservice_utils.CreateWebServiceProxy(server_url) |
| else: |
| server = server_proxy.GetServerProxy() |
| # yapf: disable |
| if self.args.raw_invocation: # type: ignore #TODO(b/338318729) Fixit! # pylint: disable=line-too-long |
| # yapf: enable |
| raise ValueError('Argument `raw_invocation` allowed only for external ' |
| 'server (need `server_url`).') |
| |
| # Prepare arguments |
| # yapf: disable |
| method = self.args.method # type: ignore #TODO(b/338318729) Fixit! # pylint: disable=line-too-long |
| # yapf: enable |
| # yapf: disable |
| args = list(self.args.args or ()) # type: ignore #TODO(b/338318729) Fixit! # pylint: disable=line-too-long |
| # yapf: enable |
| # yapf: disable |
| kargs = dict(self.args.kargs or {}) # type: ignore #TODO(b/338318729) Fixit! # pylint: disable=line-too-long |
| # yapf: enable |
| |
| # yapf: disable |
| if self.args.raw_invocation: # type: ignore #TODO(b/338318729) Fixit! # pylint: disable=line-too-long |
| # yapf: enable |
| spec = ServiceSpec(has_data=False) |
| else: |
| # yapf: disable |
| if self.args.kargs: # type: ignore #TODO(b/338318729) Fixit! # pylint: disable=line-too-long |
| # yapf: enable |
| raise ValueError('`kargs` only allowed for `raw_invocation`.') |
| # yapf: disable |
| spec = self.METHODS.get(method) # type: ignore #TODO(b/338318729) Fixit! # pylint: disable=line-too-long |
| # yapf: enable |
| if not spec: |
| raise ValueError(f'Unknown method for shopfloor service: {method}') |
| |
| if spec.data_args: |
| args = [device_data.GetDeviceData(k) for k in spec.data_args] + args |
| if spec.has_data: |
| args.insert(0, self.GetFactoryDeviceData()) |
| |
| log_args = '(...)' if spec.has_privacy_args else repr(tuple(args)) |
| log_args += repr(kargs) if kargs else '' |
| |
| logging.info('shopfloor_service: invoking %s%s', method, log_args) |
| invocation_message = pprint.pformat({method: args}) + ( |
| pprint.pformat(kargs) if kargs else '') |
| |
| # Reduce messages. |
| logger = log_utils.NoisyLogger( |
| lambda fault, prompt: logging.exception(prompt, fault)) |
| |
| while True: |
| self.ShowMessage( |
| _('Invoking shopfloor service'), 'test-status-active large', |
| invocation_message) |
| try: |
| result = getattr(server, method)(*args, **kargs) |
| logging.info('shopfloor_service: %s%s => %r', |
| method, log_args, self.FilterDict(result)) |
| self.UpdateAutoResults(method, result, args) |
| self.UpdateDeviceData(result) |
| break |
| except server_proxy.Fault as f: |
| message = f.faultString |
| logger.Log(message, 'Server fault occurred: %s') |
| self.HandleError(message, invocation_message) |
| except Exception: |
| message = debug_utils.FormatExceptionOnly() |
| logger.Log(message, 'Exception invoking shopfloor service: %s') |
| self.HandleError(message, invocation_message) |