| #!/usr/bin/env python3 |
| # -*- coding: utf-8 -*- |
| # Copyright 2018 The ChromiumOS Authors |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| """Unittests for Nebraska server.""" |
| |
| from __future__ import print_function |
| |
| import base64 |
| import builtins |
| import collections |
| import http |
| import json |
| import logging |
| import os |
| import shutil |
| import tempfile |
| import unittest |
| import unittest.mock as mock |
| from xml.etree import ElementTree |
| |
| import nebraska |
| |
| |
| _NEBRASKA_PORT = 11235 |
| _PAYLOADS_ADDRESS = "www.google.com/update/" |
| |
| # pylint: disable=protected-access |
| |
| |
| def _ExcludeNoneValuesFromDict(dictionary): |
| """Returns the input dictionary but keys with None values removed. |
| |
| Args: |
| dictionary: input dict object. |
| |
| Returns: |
| a dict object containing all the items of intput dictionary except the ones |
| which have None value. |
| """ |
| return {k: v for k, v in dictionary.items() if v is not None} |
| |
| |
| class NebraskaBaseTest(unittest.TestCase): |
| """Base class for all nebraska unittests.""" |
| |
| def setUp(self): |
| """Sets up the base test fixture.""" |
| self.tempdir = tempfile.mkdtemp() |
| |
| def tearDown(self): |
| """Cleans up the text fixture.""" |
| shutil.rmtree(self.tempdir) |
| |
| def GenerateAppData( |
| self, |
| path="foo.json", |
| appid="foo", |
| name="foo", |
| is_delta=False, |
| target_version="2.0.0", |
| source_version=None, |
| include_public_key=False, |
| ): |
| """Generates a TestAppData test instance. |
| |
| Returns: |
| The resulting data in a dictionary format. |
| """ |
| TestAppData = collections.namedtuple( |
| "TestAppData", |
| [ |
| nebraska.AppIndex.AppData.APPID_KEY, |
| nebraska.AppIndex.AppData.NAME_KEY, |
| nebraska.AppIndex.AppData.TARGET_VERSION_KEY, |
| nebraska.AppIndex.AppData.IS_DELTA_KEY, |
| nebraska.AppIndex.AppData.SOURCE_VERSION_KEY, |
| nebraska.AppIndex.AppData.SIZE_KEY, |
| nebraska.AppIndex.AppData.METADATA_SIG_KEY, |
| nebraska.AppIndex.AppData.METADATA_SIZE_KEY, |
| nebraska.AppIndex.AppData.SHA256_HEX_KEY, |
| nebraska.AppIndex.AppData.PUBLIC_KEY_RSA_KEY, |
| ], |
| ) |
| |
| data = TestAppData( |
| appid, |
| name, |
| target_version, |
| is_delta, |
| source_version, |
| "9001", |
| "Dr4RFXYgcfvFHV/0VRQs+SCQmz15Sk04LLEDswtvng8BqNbBXA7VvPUhpCgX5T/t7cwP" |
| "xTUHJVtxIREuBZpyIQxJQFZATspaClelpKBwadQzj7dpfShLpcbdlfM8qbLjIbXfC2Vy" |
| "mw03Mwf38lm0Fm75SANSTW9S4arPDf3sy9YGuqesnEJXyT3ZSGyK+Xto79zlURUXCgmi" |
| "a6t7MaJE3ZbCdeF4EiEMPDsipqj9ggmKwiCdUl2+RxTznX/015XFiBhogLrx9RCPHTR8" |
| "nLz0H9RcRhIvMd+8g4kHUZTDvjCvG5EZHpIKk8FK8z0zY/LWVr738evwuNNwyKIazkQB" |
| "TA==", |
| "42", |
| "8gBImdKWgwAkwVNCij4u1QNJqkvzOUjoGWUw8ATvjgs=", |
| # '886fd274745b4fa8d1f253cff11242fac07a29522b1bb9e028ab1480353d3160', |
| "foo-public-key" if include_public_key else None, |
| ) |
| |
| with open(os.path.join(self.tempdir, path), "w") as fp: |
| dict_version = _ExcludeNoneValuesFromDict(data._asdict()) |
| json.dump(dict_version, fp) |
| return data |
| |
| |
| def GenerateXMLAppRequest( |
| appid="foo", |
| version="1.0.0", |
| delta_okay=False, |
| track="foo-channel", |
| board="foo-board", |
| event=False, |
| event_type="1", |
| event_result="1", |
| previous_version=None, |
| update_check=True, |
| ping=False, |
| rollback_allowed=False, |
| ): |
| """Returns an XML app request.""" |
| app = ElementTree.Element( |
| "app", |
| attrib=_ExcludeNoneValuesFromDict( |
| { |
| "appid": appid, |
| "version": version, |
| "delta_okay": str(delta_okay).lower(), |
| "track": track, |
| "board": board, |
| } |
| ), |
| ) |
| |
| if ping: |
| ping = ElementTree.Element( |
| "ping", attrib={"active": "1", "a": "1", "r": "1"} |
| ) |
| app.append(ping) |
| if update_check: |
| update_check = ElementTree.Element("updatecheck") |
| if rollback_allowed: |
| update_check.set("rollback_allowed", "true") |
| app.append(update_check) |
| |
| if event: |
| event = ElementTree.Element( |
| "event", |
| attrib=_ExcludeNoneValuesFromDict( |
| { |
| "eventtype": event_type, |
| "eventresult": event_result, |
| "previousversion": previous_version, |
| } |
| ), |
| ) |
| app.append(event) |
| |
| return app |
| |
| |
| def GenerateXMLRequest(apps): |
| """Generates an XML request. |
| |
| Args: |
| apps: A list of XML app requests. |
| |
| Returns: |
| The generated XML request. |
| """ |
| root = ElementTree.Element( |
| "request", |
| attrib={ |
| "requestid": "foo-request-id", |
| "sessionid": "foo-session-id", |
| "protocol": "3.0", |
| "updater": "ChromeOSUpdateEngine", |
| "updaterversion": "0.1.0.0", |
| "installsource": "ondemandupdate", |
| "ismachine": "1", |
| }, |
| ) |
| os_tag = ElementTree.Element( |
| "os", |
| attrib={ |
| "version": "Indy", |
| "platform": "Chrome OS", |
| "sp": "12933.0.0_x86_64", |
| }, |
| ) |
| root.append(os_tag) |
| for app in apps: |
| root.append(app) |
| |
| return ElementTree.tostring(root, encoding="UTF-8", method="xml") |
| |
| |
| class MockNebraskaHandler(nebraska.NebraskaServer.NebraskaHandler): |
| """Subclass NebraskaHandler to facilitate testing. |
| |
| Because of the complexity of the socket handling super class init functions, |
| the easiest way to test NebraskaHandler is to just subclass it and mock |
| whatever we need from its super classes. |
| """ |
| |
| # pylint: disable=super-init-not-called |
| def __init__(self): |
| self.headers = mock.MagicMock() |
| self.path = mock.MagicMock() |
| self._SendResponse = mock.MagicMock() |
| self.send_error = mock.MagicMock() |
| self.rfile = mock.MagicMock() |
| self.server = mock.MagicMock() |
| nebraska_obj = nebraska.Nebraska() |
| nebraska_obj.UpdateConfig(update_payloads_address=_PAYLOADS_ADDRESS) |
| self.server.owner = nebraska.NebraskaServer(nebraska_obj) |
| |
| |
| class NebraskaHandlerTest(unittest.TestCase): |
| """Test NebraskaHandler.""" |
| |
| # TODO(ahasssni): Change these tests to be like tests in NebraskaTest. |
| def testParseURL(self): |
| """Tests _ParseURL with different URLs.""" |
| nebraska_handler = MockNebraskaHandler() |
| self.assertEqual( |
| nebraska_handler._ParseURL("http://goo.gle/path/?tick=tock"), |
| ("path", {"tick": ["tock"]}), |
| ) |
| |
| self.assertEqual( |
| nebraska_handler._ParseURL("http://goo.gle/path/"), ("path", {}) |
| ) |
| |
| self.assertEqual( |
| nebraska_handler._ParseURL("http://goo.gle/"), ("", {}) |
| ) |
| |
| @mock.patch.object( |
| nebraska.Nebraska, "GetResponseToRequest", return_value="foobar" |
| ) |
| @mock.patch.object(nebraska, "Request") |
| def testDoPostSuccess(self, _, response_mock): |
| """Tests do_POST success.""" |
| nebraska_handler = MockNebraskaHandler() |
| nebraska_handler.path = "http://test.com/update" |
| |
| nebraska_handler.do_POST() |
| |
| response_mock.assert_called_once() |
| nebraska_handler._SendResponse.assert_called_once_with( |
| "application/xml", response_mock.return_value |
| ) |
| |
| @mock.patch.object(nebraska.Nebraska, "GetResponseToRequest") |
| @mock.patch.object(nebraska, "Request") |
| def testDoPostSuccessWithCriticalUpdate(self, _, response_mock): |
| """Tests do_POST success with critical_update query string in URL.""" |
| nebraska_handler = MockNebraskaHandler() |
| nebraska_handler.path = "http://test.com/update/?critical_update=True" |
| |
| nebraska_handler.do_POST() |
| |
| response_mock.assert_called_once() |
| self.assertTrue( |
| nebraska_handler.server.owner.nebraska._config.critical_update |
| ) |
| |
| @mock.patch.object(nebraska.Nebraska, "GetResponseToRequest") |
| @mock.patch.object(nebraska, "Request") |
| def testDoPostSuccessWithNoUpdate(self, _, response_mock): |
| """Tests do_POST success with no_update query string in URL.""" |
| nebraska_handler = MockNebraskaHandler() |
| nebraska_handler.path = "http://test.com/update/?no_update=True" |
| |
| nebraska_handler.do_POST() |
| |
| response_mock.assert_called_once() |
| self.assertTrue( |
| nebraska_handler.server.owner.nebraska._config.no_update |
| ) |
| |
| def testDoPostInvalidPath(self): |
| """Test do_POST invalid path.""" |
| nebraska_handler = MockNebraskaHandler() |
| nebraska_handler.path = "http://test.com/invalid-path" |
| nebraska_handler.do_POST() |
| |
| nebraska_handler.send_error.assert_called_once_with( |
| http.client.BAD_REQUEST, |
| 'The requested path "invalid-path" was not found!', |
| ) |
| |
| @mock.patch.object(nebraska, "traceback") |
| @mock.patch.object( |
| nebraska.Request, |
| "ParseRequest", |
| side_effect=nebraska.InvalidRequestError, |
| ) |
| def testDoPostInvalidRequest(self, _, traceback_mock): |
| """Test do_POST invalid request.""" |
| nebraska_handler = MockNebraskaHandler() |
| nebraska_handler.path = "http://test.com/update" |
| nebraska_handler.do_POST() |
| |
| self.assertEqual(traceback_mock.format_exc.call_count, 2) |
| nebraska_handler.send_error.assert_called_once_with( |
| http.client.INTERNAL_SERVER_ERROR, traceback_mock.format_exc() |
| ) |
| |
| @mock.patch.object(nebraska, "traceback") |
| @mock.patch.object(nebraska, "Response") |
| def testDoPostInvalidResponse(self, response_mock, traceback_mock): |
| """Tests do_POST invalid response handling.""" |
| nebraska_handler = MockNebraskaHandler() |
| nebraska_handler.path = "http://test.com/update" |
| |
| response_instance = response_mock.return_value |
| response_instance.GetXMLString.side_effect = Exception |
| nebraska_handler.do_POST() |
| |
| self.assertEqual(traceback_mock.format_exc.call_count, 2) |
| nebraska_handler.send_error.assert_called_once_with( |
| http.client.INTERNAL_SERVER_ERROR, traceback_mock.format_exc() |
| ) |
| |
| def testDoPostUpdateConfig(self): |
| """Tests do_POST success for update_config API.""" |
| nebraska_handler = MockNebraskaHandler() |
| nebraska_handler.path = "http://test.com/update_config" |
| nebraska_handler.rfile.read = lambda x: json.dumps( |
| {"critical_update": True} |
| ) |
| |
| self.assertFalse( |
| nebraska_handler.server.owner.nebraska._config.critical_update |
| ) |
| |
| nebraska_handler.do_POST() |
| |
| self.assertTrue( |
| nebraska_handler.server.owner.nebraska._config.critical_update |
| ) |
| nebraska_handler._SendResponse.assert_called_once_with( |
| "text/plain", b"Config set!" |
| ) |
| |
| def testDoGetFailureBadPath(self): |
| """Tests do_GET failure on bad path.""" |
| nebraska_handler = MockNebraskaHandler() |
| nebraska_handler.path = "http://test.com/invalid-path" |
| |
| nebraska_handler.do_GET() |
| nebraska_handler.send_error(http.client.BAD_REQUEST, mock.ANY) |
| |
| def testDoGetHealthCheck(self): |
| """Tests do_GET with health_check path.""" |
| nebraska_handler = MockNebraskaHandler() |
| nebraska_handler.path = "http://test.com/health_check" |
| |
| nebraska_handler.do_GET() |
| nebraska_handler._SendResponse.assert_called_once_with( |
| "text/plain", b"Nebraska is alive!" |
| ) |
| |
| |
| class NebraskaServerTest(NebraskaBaseTest): |
| """Test NebraskaServer.""" |
| |
| def testStart(self): |
| """Tests start of server.""" |
| nebraska_instance = nebraska.Nebraska() |
| nebraska_instance.UpdateConfig( |
| update_payloads_address=_PAYLOADS_ADDRESS |
| ) |
| server = nebraska.NebraskaServer(nebraska_instance, port=_NEBRASKA_PORT) |
| |
| with mock.patch.object(http.server, "HTTPServer") as server_mock: |
| with mock.patch.object(nebraska.threading, "Thread") as thread_mock: |
| server.Start() |
| |
| server_mock.assert_called_once_with( |
| ("", _NEBRASKA_PORT), |
| nebraska.NebraskaServer.NebraskaHandler, |
| ) |
| |
| # pylint: disable=protected-access |
| thread_mock.assert_has_calls( |
| [ |
| mock.call(target=server._httpd.serve_forever), |
| mock.call().start(), |
| ] |
| ) |
| |
| def testStop(self): |
| """Tests Stop.""" |
| nebraska_instance = nebraska.Nebraska() |
| nebraska_instance.UpdateConfig( |
| update_payloads_address=_PAYLOADS_ADDRESS |
| ) |
| server = nebraska.NebraskaServer(nebraska_instance, port=_NEBRASKA_PORT) |
| |
| # pylint: disable=protected-access |
| server._httpd = mock.MagicMock(name="_httpd") |
| server._server_thread = mock.MagicMock(name="_server_thread") |
| server.Stop() |
| # pylint: disable=protected-access |
| server._httpd.shutdown.assert_called_once_with() |
| server._server_thread.join.assert_called_once_with() |
| |
| def testMiscFiles(self): |
| """Tests PID and port files are correctly written.""" |
| temp_dir = tempfile.mkdtemp() |
| runtime_root = os.path.join(temp_dir, "runtime_root") |
| nebraska_instance = nebraska.Nebraska() |
| nebraska_instance.UpdateConfig( |
| update_payloads_address=_PAYLOADS_ADDRESS |
| ) |
| server = nebraska.NebraskaServer( |
| nebraska_instance, port=_NEBRASKA_PORT, runtime_root=runtime_root |
| ) |
| |
| port_file = os.path.join(runtime_root, "port") |
| pid_file = os.path.join(runtime_root, "pid") |
| |
| with mock.patch.object(http.server, "HTTPServer"): |
| with mock.patch.object(nebraska.threading, "Thread"): |
| server.Start() |
| |
| # Make sure files are created and written with correct values. |
| with open(port_file, "r") as f: |
| self.assertEqual(f.read(), str(server.GetPort())) |
| with open(pid_file, "r") as f: |
| self.assertEqual(f.read(), str(os.getpid())) |
| |
| server.Stop() |
| |
| # Make sure files are deleted correctly. |
| self.assertFalse(os.path.exists(runtime_root)) |
| |
| # Delete the temp directory. |
| shutil.rmtree(temp_dir, ignore_errors=True) |
| |
| |
| class NebraskaTest(NebraskaBaseTest): |
| """Test AppIndex.""" |
| |
| def testDefaultInstallPayloadsAddress(self): |
| """Tests the default install_payloads_address is correctly set.""" |
| update_addr = "foo/update/" |
| install_addr = "foo/install/" |
| n = nebraska.Nebraska() |
| n.UpdateConfig( |
| update_payloads_address=update_addr, |
| install_payloads_address=install_addr, |
| ) |
| # pylint: disable=protected-access |
| self.assertEqual(n._config.install_payloads_address, install_addr) |
| self.assertEqual(n._config.update_payloads_address, update_addr) |
| |
| def testQueryDictToDict(self): |
| """Tests QueryDictToDict() function""" |
| self.assertEqual( |
| nebraska.QueryDictToDict( |
| { |
| "critical_update": "True", |
| "disable_payload_backoff": "False", |
| "eol_date": "10", |
| "failures_per_url": "1", |
| "no_update": "True", |
| "num_urls": ["0", "1"], |
| "foo": "bar", |
| } |
| ), |
| { |
| "critical_update": True, |
| "disable_payload_backoff": False, |
| "eol_date": 10, |
| "failures_per_url": 1, |
| "no_update": True, |
| "num_urls": 0, |
| }, |
| ) |
| |
| def testDefaultRequestResponseXML(self): |
| """Tests the default response to a request for an update check.""" |
| app_data = self.GenerateAppData() |
| neb = nebraska.Nebraska() |
| neb.UpdateConfig( |
| update_app_index=nebraska.AppIndex(self.tempdir), |
| update_payloads_address=_PAYLOADS_ADDRESS, |
| ) |
| request = GenerateXMLRequest([GenerateXMLAppRequest()]) |
| response = neb.GetResponseToRequest(nebraska.Request(request)) |
| |
| root = ElementTree.fromstring(response) |
| apps = root.findall("app") |
| self.assertEqual(len(apps), 1) |
| self.assertEqual(apps[0].attrib["appid"], app_data.appid) |
| self.assertEqual(apps[0].attrib["status"], "ok") |
| |
| update_check = apps[0].find("updatecheck") |
| self.assertEqual(update_check.attrib["status"], "ok") |
| |
| self.assertNotIn("_is_rollback", update_check.attrib) |
| index_strs = ["", "_0", "_1", "_2", "_3", "_4"] |
| for idx in index_strs: |
| self.assertEqual( |
| update_check.attrib["_firmware_version" + idx], "1.1" |
| ) |
| self.assertEqual( |
| update_check.attrib["_kernel_version" + idx], "1.1" |
| ) |
| |
| self.assertNotIn("_eol_date", update_check.attrib) |
| |
| self.assertNotIn("_extended_date", update_check.attrib) |
| self.assertNotIn("_extended_opt_in_required", update_check.attrib) |
| |
| urls = update_check.findall("urls/url") |
| self.assertEqual(len(urls), 1) |
| self.assertEqual(urls[0].attrib["codebase"], _PAYLOADS_ADDRESS) |
| |
| manifest = update_check.find("manifest") |
| self.assertIsNotNone(manifest) |
| self.assertEqual(manifest.attrib["version"], app_data.target_version) |
| |
| actions = manifest.findall("actions/action") |
| self.assertEqual(len(actions), 2) |
| self.assertEqual(actions[1].attrib["sha256"], app_data.sha256_hex) |
| self.assertEqual( |
| actions[1].attrib["ChromeOSVersion"], app_data.target_version |
| ) |
| self.assertEqual(actions[1].attrib["DisablePayloadBackoff"], "false") |
| self.assertEqual( |
| actions[1].attrib["MetadataSignatureRsa"], |
| app_data.metadata_signature, |
| ) |
| self.assertEqual( |
| actions[1].attrib["MetadataSize"], app_data.metadata_size |
| ) |
| self.assertNotIn("MaxFailureCountPerUrl", actions[1].attrib) |
| self.assertEqual( |
| actions[1].attrib["ChromeOSVersion"], app_data.target_version |
| ) |
| self.assertEqual( |
| actions[1].attrib["IsDeltaPayload"], str(app_data.is_delta).lower() |
| ) |
| self.assertNotIn("deadline", actions[1].attrib) |
| self.assertNotIn("PublicKeyRsa", actions[1].attrib) |
| self.assertNotIn("MaxFailureCountPerUrl", actions[1].attrib) |
| |
| package = manifest.find("packages/package") |
| self.assertIsNotNone(package) |
| self.assertEqual(package.attrib["name"], app_data.name) |
| self.assertEqual(package.attrib["size"], app_data.size) |
| sha256_hex = base64.b16encode( |
| base64.b64decode(app_data.sha256_hex) |
| ).decode("utf-8") |
| self.assertEqual(package.attrib["hash_sha256"], sha256_hex) |
| self.assertEqual(package.attrib["fp"], "1.%s" % sha256_hex) |
| |
| def testScanEmpty(self): |
| """Tests Scan on an empty directory.""" |
| with mock.patch("os.listdir") as listdir_mock: |
| with mock.patch.object(builtins, "open") as open_mock: |
| listdir_mock.return_value = [] |
| app_index = nebraska.AppIndex("foo-dir") |
| self.assertFalse(app_index._index) |
| listdir_mock.assert_called_once_with("foo-dir") |
| open_mock.assert_not_called() |
| |
| def testScanNoJson(self): |
| """Tests Scan on a directory with no JSON files.""" |
| with mock.patch("os.listdir") as listdir_mock: |
| with mock.patch.object(builtins, "open") as open_mock: |
| listdir_mock.return_value = ["foo.bin", "bar.bin", "json"] |
| app_index = nebraska.AppIndex("foo-dir") |
| self.assertFalse(app_index._index) |
| listdir_mock.assert_called_once_with("foo-dir") |
| open_mock.assert_not_called() |
| |
| def testScanMultiple(self): |
| """Tests Scan on a directory with multiple appids.""" |
| # Providing some mock properties and non-properties files. |
| self.GenerateAppData("foo_update.json") |
| self.GenerateAppData( |
| "foo_install.json", is_delta=True, source_version="foo-version" |
| ) |
| self.GenerateAppData("bar_install.json", appid="bar") |
| self.GenerateAppData( |
| "bar_update.json", |
| appid="bar", |
| is_delta=True, |
| source_version="foo-version", |
| ) |
| self.GenerateAppData("foo.blah", appid="foobar") |
| |
| # Make sure the Scan() scans all the files and at least correct App IDs |
| # are generated. 'foo' and 'bar' App IDs should appear twice each. |
| app_index = nebraska.AppIndex(self.tempdir) |
| expected_appids = [x.appid for x in app_index._index] |
| self.assertEqual(len(expected_appids), 4) |
| self.assertEqual(expected_appids.count("foo"), 2) |
| self.assertEqual(expected_appids.count("bar"), 2) |
| |
| def testScanInvalidJson(self): |
| """Tests Scan with invalid JSON files.""" |
| self.GenerateAppData("foo_update.json") |
| with mock.patch.object(builtins, "open", side_effect=IOError): |
| # Make sure we raise error when loading files raises one. |
| with self.assertRaises(IOError): |
| nebraska.AppIndex(self.tempdir) |
| |
| def testScanInvalidApp(self): |
| """Tests Scan on JSON files lacking required keys.""" |
| self.GenerateAppData("foo_update.json", appid=None) |
| # Make sure we raise error when properties files are invalid. |
| with self.assertRaises(KeyError): |
| nebraska.AppIndex(self.tempdir) |
| |
| def testMatch(self): |
| """Tests different scenarios for correctly matching AppData.""" |
| # Providing some properties files. |
| self.GenerateAppData("foo_install.json") |
| neb = nebraska.Nebraska() |
| neb.UpdateConfig( |
| update_app_index=nebraska.AppIndex(self.tempdir), |
| update_payloads_address=_PAYLOADS_ADDRESS, |
| ) |
| |
| # Nothing found. |
| request = GenerateXMLRequest([GenerateXMLAppRequest(appid="random")]) |
| response = neb.GetResponseToRequest(nebraska.Request(request)) |
| root = ElementTree.fromstring(response) |
| app = root.find("app") |
| self.assertEqual(app.attrib["appid"], "random") |
| update_check = app.find("updatecheck") |
| self.assertEqual(update_check.attrib["status"], "noupdate") |
| |
| # Partially matches against the AppData with appid 'foo'. |
| request = GenerateXMLRequest([GenerateXMLAppRequest(appid="mefoolme")]) |
| response = neb.GetResponseToRequest(nebraska.Request(request)) |
| root = ElementTree.fromstring(response) |
| app = root.find("app") |
| self.assertEqual(app.attrib["appid"], "mefoolme") |
| update_check = app.find("updatecheck") |
| self.assertEqual(update_check.attrib["status"], "ok") |
| |
| def testMatchEmpty(self): |
| """Tests Constains() correctly finds matching AppData with empty appid.""" |
| self.GenerateAppData("foo_update.json", appid="") |
| neb = nebraska.Nebraska() |
| neb.UpdateConfig( |
| update_app_index=nebraska.AppIndex(self.tempdir), |
| update_payloads_address=_PAYLOADS_ADDRESS, |
| ) |
| request = GenerateXMLRequest([GenerateXMLAppRequest()]) |
| response = neb.GetResponseToRequest(nebraska.Request(request)) |
| |
| root = ElementTree.fromstring(response) |
| app = root.find("app") |
| self.assertEqual(app.attrib["appid"], "foo") |
| update_check = app.find("updatecheck") |
| self.assertEqual(update_check.attrib["status"], "ok") |
| |
| def testMatchInstall(self): |
| """Tests matching install requests.""" |
| self.GenerateAppData("foo.json") |
| self.GenerateAppData("bar.json", appid="bar") |
| neb = nebraska.Nebraska() |
| neb.UpdateConfig( |
| install_app_index=nebraska.AppIndex(self.tempdir), |
| install_payloads_address=_PAYLOADS_ADDRESS, |
| ) |
| request = GenerateXMLRequest( |
| [ |
| # Not having an updatecheck tag in the platform App causes this to be an |
| # install request. |
| GenerateXMLAppRequest(appid="foo", update_check=False), |
| GenerateXMLAppRequest(appid="bar", version="0.0.0.0"), |
| ] |
| ) |
| response = neb.GetResponseToRequest(nebraska.Request(request)) |
| |
| apps = ElementTree.fromstring(response).findall("app") |
| self.assertEqual(apps[0].attrib["status"], "noupdate") |
| update_check = apps[1].find("updatecheck") |
| self.assertEqual(update_check.attrib["status"], "ok") |
| |
| def testMatchDeltaAndFull(self): |
| """Tests matching delta update request.""" |
| # Two properties file with the same App ID but one delta, one full. |
| self.GenerateAppData("foo.json") |
| self.GenerateAppData("bar.json", is_delta=True, source_version="1.0.0") |
| neb = nebraska.Nebraska() |
| neb.UpdateConfig( |
| update_app_index=nebraska.AppIndex(self.tempdir), |
| update_payloads_address=_PAYLOADS_ADDRESS, |
| ) |
| |
| # Full payload. |
| request = GenerateXMLRequest([GenerateXMLAppRequest(delta_okay=False)]) |
| response = neb.GetResponseToRequest(nebraska.Request(request)) |
| root = ElementTree.fromstring(response) |
| action = root.findall("app/updatecheck/manifest/actions/action")[1] |
| self.assertEqual(action.attrib["IsDeltaPayload"], "false") |
| |
| # Delta payload. |
| request = GenerateXMLRequest([GenerateXMLAppRequest(delta_okay=True)]) |
| response = neb.GetResponseToRequest(nebraska.Request(request)) |
| root = ElementTree.fromstring(response) |
| action = root.findall("app/updatecheck/manifest/actions/action")[1] |
| self.assertEqual(action.attrib["IsDeltaPayload"], "true") |
| |
| # Forced full payload should return full payload. |
| neb.UpdateConfig(full_payload=True) |
| response = neb.GetResponseToRequest(nebraska.Request(request)) |
| root = ElementTree.fromstring(response) |
| action = root.findall("app/updatecheck/manifest/actions/action")[1] |
| self.assertEqual(action.attrib["IsDeltaPayload"], "false") |
| |
| # Delta payload should return delta payload. |
| neb.UpdateConfig(full_payload=False) |
| response = neb.GetResponseToRequest(nebraska.Request(request)) |
| root = ElementTree.fromstring(response) |
| action = root.findall("app/updatecheck/manifest/actions/action")[1] |
| self.assertEqual(action.attrib["IsDeltaPayload"], "true") |
| |
| # Forced delta payload when client can't get delta should be noupdate. |
| request = GenerateXMLRequest([GenerateXMLAppRequest(delta_okay=False)]) |
| response = neb.GetResponseToRequest(nebraska.Request(request)) |
| root = ElementTree.fromstring(response) |
| update_check = root.find("app/updatecheck") |
| self.assertEqual(update_check.attrib["status"], "noupdate") |
| |
| def testMismatchAppDataAppID(self): |
| """Tests appid mismatch.""" |
| self.GenerateAppData(appid="bar") |
| neb = nebraska.Nebraska() |
| neb.UpdateConfig( |
| update_app_index=nebraska.AppIndex(self.tempdir), |
| update_payloads_address=_PAYLOADS_ADDRESS, |
| ) |
| request = GenerateXMLRequest([GenerateXMLAppRequest(appid="foo")]) |
| response = neb.GetResponseToRequest(nebraska.Request(request)) |
| |
| update_check = ElementTree.fromstring(response).find("app/updatecheck") |
| self.assertEqual(update_check.attrib["status"], "noupdate") |
| |
| def testIgnoreAppDataAppID(self): |
| """Tests ignoring appid.""" |
| self.GenerateAppData(appid="bar") |
| neb = nebraska.Nebraska() |
| neb.UpdateConfig( |
| update_app_index=nebraska.AppIndex(self.tempdir), |
| update_payloads_address=_PAYLOADS_ADDRESS, |
| ignore_appid=True, |
| ) |
| request = GenerateXMLRequest([GenerateXMLAppRequest(appid="foo")]) |
| response = neb.GetResponseToRequest(nebraska.Request(request)) |
| |
| update_check = ElementTree.fromstring(response).find("app/updatecheck") |
| self.assertEqual(update_check.attrib["status"], "ok") |
| |
| def testMatchCanaryAppId(self): |
| """Tests matching update request with canary appid.""" |
| self.GenerateAppData(appid="1" * len(nebraska._CANARY_APP_ID) + "foo") |
| neb = nebraska.Nebraska() |
| neb.UpdateConfig( |
| update_app_index=nebraska.AppIndex(self.tempdir), |
| update_payloads_address=_PAYLOADS_ADDRESS, |
| ) |
| request = GenerateXMLRequest( |
| [GenerateXMLAppRequest(appid=nebraska._CANARY_APP_ID + "foo")] |
| ) |
| response = neb.GetResponseToRequest(nebraska.Request(request)) |
| |
| update_check = ElementTree.fromstring(response).find("app/updatecheck") |
| self.assertEqual(update_check.attrib["status"], "ok") |
| |
| def testAlreadyMatched(self): |
| """Tests not matching an already matched app.""" |
| self.GenerateAppData() |
| neb = nebraska.Nebraska() |
| neb.UpdateConfig( |
| update_app_index=nebraska.AppIndex(self.tempdir), |
| update_payloads_address=_PAYLOADS_ADDRESS, |
| ) |
| |
| # Two identical requests. |
| request = GenerateXMLRequest( |
| [GenerateXMLAppRequest(), GenerateXMLAppRequest()] |
| ) |
| response = neb.GetResponseToRequest(nebraska.Request(request)) |
| update_checks = ElementTree.fromstring(response).findall( |
| "app/updatecheck" |
| ) |
| self.assertEqual(update_checks[0].attrib["status"], "ok") |
| self.assertEqual(update_checks[1].attrib["status"], "noupdate") |
| |
| def testInvalidXMLRequest(self): |
| """Tests ParseRequest() handling of invalid XML.""" |
| with self.assertRaises(nebraska.InvalidRequestError): |
| nebraska.Request("invalid xml!") |
| |
| def testInvalidAppRequest(self): |
| """Tests ParseRequest handling of invalid app requests.""" |
| request = GenerateXMLRequest([GenerateXMLAppRequest(appid=None)]) |
| with self.assertRaises(nebraska.InvalidRequestError): |
| nebraska.Request(request) |
| |
| def testParseRequestInvalidInstall(self): |
| """Tests handling of invalid install app request (missing version).""" |
| request = GenerateXMLRequest( |
| [ |
| GenerateXMLAppRequest(appid="foo", update_check=None), |
| GenerateXMLAppRequest(appid="test", version=None), |
| ] |
| ) |
| with self.assertRaises(nebraska.InvalidRequestError): |
| nebraska.Request(request) |
| |
| def testParseRequestInvalidNoop(self): |
| """Tests ParseRequest handling of invalid mixed no-op request.""" |
| request = GenerateXMLRequest( |
| [ |
| GenerateXMLAppRequest(update_check=None), |
| GenerateXMLAppRequest(appid="bar"), |
| GenerateXMLAppRequest(appid="test", update_check=None), |
| ] |
| ) |
| with self.assertRaises(nebraska.InvalidRequestError): |
| nebraska.Request(request) |
| |
| def testParseRequestMissingAtLeastOneRequiredAttr(self): |
| """Tests ParseRequest handling of missing required attributes in request.""" |
| request = GenerateXMLRequest( |
| [ |
| GenerateXMLAppRequest(track=None), |
| GenerateXMLAppRequest(appid="bar", track=None), |
| ] |
| ) |
| with self.assertRaises(nebraska.InvalidRequestError): |
| nebraska.Request(request) |
| |
| def testParseRequestMismatchedVersionUpdate(self): |
| """Tests ParseRequest handling of mismatched update version numbers.""" |
| request = GenerateXMLRequest( |
| [ |
| GenerateXMLAppRequest(version="1.0.0"), |
| GenerateXMLAppRequest(appid="bar", version="2.0.0"), |
| ] |
| ) |
| with self.assertRaises(nebraska.InvalidRequestError): |
| nebraska.Request(request) |
| |
| def testParseRequestMiniOSMismatchedVersion(self): |
| """Tests ParseRequest handling of miniOS mismtach update.""" |
| request = GenerateXMLRequest( |
| [ |
| GenerateXMLAppRequest(version="1.0.0"), |
| GenerateXMLAppRequest( |
| appid="foo" + nebraska._MINIOS_APP_ID_SUFFIX, |
| version="2.0.0", |
| ), |
| ] |
| ) |
| # Should not assert. |
| nebraska.Request(request) |
| |
| def testParseRequestMismatchedVersionInstall(self): |
| """Tests ParseRequest handling of mismatched install version numbers.""" |
| request = GenerateXMLRequest( |
| [ |
| GenerateXMLAppRequest(version="1.0.0", update_check=None), |
| GenerateXMLAppRequest(appid="bar", version="2.0.0"), |
| ] |
| ) |
| with self.assertRaises(nebraska.InvalidRequestError): |
| nebraska.Request(request) |
| |
| def testUpdateMultipleApps(self): |
| """Tests update with multiple apps.""" |
| app_datas = [ |
| self.GenerateAppData("foo.json", appid="foo"), |
| self.GenerateAppData("bar.json", appid="bar"), |
| self.GenerateAppData("test.json", appid="test"), |
| ] |
| neb = nebraska.Nebraska() |
| neb.UpdateConfig( |
| update_app_index=nebraska.AppIndex(self.tempdir), |
| update_payloads_address=_PAYLOADS_ADDRESS, |
| ) |
| request = GenerateXMLRequest( |
| [ |
| GenerateXMLAppRequest(appid="foo"), |
| GenerateXMLAppRequest(appid="bar"), |
| GenerateXMLAppRequest(appid="test"), |
| ] |
| ) |
| |
| request = nebraska.Request(request) |
| self.assertEqual(request.version, "1.0.0") |
| |
| response = neb.GetResponseToRequest(request) |
| root = ElementTree.fromstring(response) |
| apps = root.findall("app") |
| self.assertEqual(len(app_datas), len(apps)) |
| for i, app in enumerate(apps): |
| self.assertEqual(app.attrib["appid"], app_datas[i].appid) |
| |
| update_checks = root.findall("app/updatecheck") |
| for i, update_check in enumerate(update_checks): |
| self.assertEqual(update_check.attrib["status"], "ok") |
| manifest = update_check.find("manifest") |
| self.assertEqual( |
| manifest.attrib["version"], app_datas[i].target_version |
| ) |
| |
| def testInstallMultipleApps(self): |
| """Tests install for multiple apps.""" |
| app_datas = [ |
| self.GenerateAppData("foo.json", appid="foo"), |
| self.GenerateAppData("bar.json", appid="bar"), |
| self.GenerateAppData("test.json", appid="test"), |
| ] |
| neb = nebraska.Nebraska() |
| neb.UpdateConfig( |
| install_app_index=nebraska.AppIndex(self.tempdir), |
| install_payloads_address=_PAYLOADS_ADDRESS, |
| ) |
| request = GenerateXMLRequest( |
| [ |
| GenerateXMLAppRequest( |
| appid="foo", version="1.0.0", update_check=None |
| ), |
| GenerateXMLAppRequest(appid="bar", version="0.0.0.0"), |
| GenerateXMLAppRequest(appid="test", version="0.0.0.0"), |
| ] |
| ) |
| |
| request = nebraska.Request(request) |
| self.assertEqual(request.version, "1.0.0") |
| |
| response = neb.GetResponseToRequest(request) |
| root = ElementTree.fromstring(response) |
| apps = root.findall("app") |
| self.assertEqual(len(app_datas), len(apps)) |
| for i, app in enumerate(apps): |
| self.assertEqual(app.attrib["appid"], app_datas[i].appid) |
| |
| update_checks = root.findall("app/updatecheck") |
| self.assertEqual(update_checks[0].attrib["status"], "ok") |
| # Only the last two apps have updatechecks. |
| for i in range(len(apps) - 1): |
| manifest = update_checks[i].find("manifest") |
| self.assertEqual( |
| manifest.attrib["version"], app_datas[i + 1].target_version |
| ) |
| |
| def testReturnNoUpdateStarting(self): |
| """Tests return_noupdate_starting with multiple apps.""" |
| self.GenerateAppData("foo.json", appid="foo") |
| self.GenerateAppData("bar.json", appid="bar") |
| neb = nebraska.Nebraska() |
| neb.UpdateConfig( |
| update_app_index=nebraska.AppIndex(self.tempdir), |
| update_payloads_address=_PAYLOADS_ADDRESS, |
| ) |
| request = GenerateXMLRequest( |
| [ |
| GenerateXMLAppRequest(appid="foo"), |
| GenerateXMLAppRequest(appid="bar"), |
| ] |
| ) |
| |
| def TestHasUpdate(has_update_check): |
| response = neb.GetResponseToRequest(nebraska.Request(request)) |
| root = ElementTree.fromstring(response) |
| update_checks = root.findall("app/updatecheck") |
| for update_check in update_checks: |
| self.assertEqual( |
| update_check.attrib["status"], |
| "ok" if has_update_check else "noupdate", |
| ) |
| |
| # With default value (0), it should always return an update. |
| TestHasUpdate(True) |
| TestHasUpdate(True) |
| TestHasUpdate(True) |
| |
| # With value 1, it should start returning noupdate from the beginning. |
| neb.UpdateConfig(return_noupdate_starting=1) |
| TestHasUpdate(False) |
| |
| # With value N greater than 1, it should return update for first N-1. |
| neb.UpdateConfig(return_noupdate_starting=3, no_update=False) |
| TestHasUpdate(True) |
| TestHasUpdate(True) |
| TestHasUpdate(False) |
| |
| def testInvalidateLastUpdate(self): |
| """Tests invalidate_last_update.""" |
| self.GenerateAppData("foo.json", appid="foo") |
| neb = nebraska.Nebraska() |
| neb.UpdateConfig( |
| update_app_index=nebraska.AppIndex(self.tempdir), |
| update_payloads_address=_PAYLOADS_ADDRESS, |
| invalidate_last_update=True, |
| ) |
| request = GenerateXMLRequest( |
| [ |
| GenerateXMLAppRequest(appid="foo"), |
| ] |
| ) |
| response = neb.GetResponseToRequest(nebraska.Request(request)) |
| app = ElementTree.fromstring(response).find("app") |
| self.assertEqual(app.attrib["appid"], "foo") |
| update_check = app.find("updatecheck") |
| self.assertEqual(update_check.attrib["status"], "noupdate") |
| self.assertEqual(update_check.attrib["_invalidate_last_update"], "true") |
| |
| def testEvent(self): |
| """Tests event requests.""" |
| neb = nebraska.Nebraska() |
| neb.UpdateConfig( |
| update_app_index=nebraska.AppIndex(self.tempdir), |
| update_payloads_address=_PAYLOADS_ADDRESS, |
| ) |
| request = GenerateXMLRequest( |
| [ |
| GenerateXMLAppRequest( |
| appid="foo", event=True, previous_version="1" |
| ), |
| GenerateXMLAppRequest( |
| appid="test", event=True, previous_version="1" |
| ), |
| ] |
| ) |
| |
| request = nebraska.Request(request) |
| self.assertEqual(request.version, "1.0.0") |
| |
| response = neb.GetResponseToRequest(request) |
| root = ElementTree.fromstring(response) |
| apps = root.findall("app") |
| self.assertEqual(len(apps), 2) |
| for app in apps: |
| event = app.find("event") |
| self.assertEqual(event.attrib["status"], "ok") |
| |
| def testPing(self): |
| """Tests ping requests.""" |
| neb = nebraska.Nebraska() |
| neb.UpdateConfig( |
| update_app_index=nebraska.AppIndex(self.tempdir), |
| update_payloads_address=_PAYLOADS_ADDRESS, |
| ) |
| request = GenerateXMLRequest( |
| [ |
| GenerateXMLAppRequest(appid="foo", ping=True), |
| GenerateXMLAppRequest(appid="test", ping=True), |
| ] |
| ) |
| |
| request = nebraska.Request(request) |
| self.assertEqual(request.version, "1.0.0") |
| |
| response = neb.GetResponseToRequest(request) |
| root = ElementTree.fromstring(response) |
| |
| apps = root.findall("app") |
| self.assertEqual(len(apps), 2) |
| for app in apps: |
| ping = app.find("ping") |
| self.assertEqual(ping.attrib["status"], "ok") |
| |
| def testCriticalUpdate(self): |
| """Tests correct response for critical updates.""" |
| self.GenerateAppData() |
| neb = nebraska.Nebraska() |
| neb.UpdateConfig( |
| update_app_index=nebraska.AppIndex(self.tempdir), |
| update_payloads_address=_PAYLOADS_ADDRESS, |
| ) |
| request = GenerateXMLRequest([GenerateXMLAppRequest()]) |
| |
| neb.UpdateConfig(critical_update=True) |
| response = neb.GetResponseToRequest(nebraska.Request(request)) |
| root = ElementTree.fromstring(response) |
| action_tag = root.findall("app/updatecheck/manifest/actions/action")[1] |
| self.assertEqual(action_tag.attrib["deadline"], "now") |
| |
| def testPublicKey(self): |
| """Tests public key is included in the response.""" |
| app_data = self.GenerateAppData(include_public_key=True) |
| neb = nebraska.Nebraska() |
| neb.UpdateConfig( |
| update_app_index=nebraska.AppIndex(self.tempdir), |
| update_payloads_address=_PAYLOADS_ADDRESS, |
| ) |
| request = GenerateXMLRequest([GenerateXMLAppRequest()]) |
| |
| response = neb.GetResponseToRequest(nebraska.Request(request)) |
| root = ElementTree.fromstring(response) |
| action_tag = root.findall("app/updatecheck/manifest/actions/action")[1] |
| self.assertEqual(action_tag.attrib["PublicKeyRsa"], app_data.public_key) |
| |
| def testRollback(self): |
| """Tests rollback parametes are setup correctly.""" |
| self.GenerateAppData() |
| neb = nebraska.Nebraska() |
| neb.UpdateConfig( |
| update_app_index=nebraska.AppIndex(self.tempdir), |
| update_payloads_address=_PAYLOADS_ADDRESS, |
| ) |
| request = GenerateXMLRequest( |
| [GenerateXMLAppRequest(rollback_allowed=True)] |
| ) |
| |
| neb.UpdateConfig(is_rollback=True) |
| neb.UpdateConfig(rollback_prevention_image=["1.1", "1.2"]) |
| neb.UpdateConfig( |
| rollback_prevention_stable=[ |
| ["1.3", "1.4"], |
| ["1.5", "1.6"], |
| ["1.7", "1.8"], |
| ["2.3", "3.4"], |
| ["4.3", "5.4"], |
| ] |
| ) |
| response = neb.GetResponseToRequest(nebraska.Request(request)) |
| root = ElementTree.fromstring(response) |
| update_check_tag = root.find("app/updatecheck") |
| |
| self.assertEqual(update_check_tag.attrib["_rollback"], "true") |
| self.assertEqual(update_check_tag.attrib["_firmware_version"], "1.1") |
| self.assertEqual(update_check_tag.attrib["_kernel_version"], "1.2") |
| self.assertEqual(update_check_tag.attrib["_firmware_version_0"], "1.3") |
| self.assertEqual(update_check_tag.attrib["_kernel_version_0"], "1.4") |
| self.assertEqual(update_check_tag.attrib["_firmware_version_1"], "1.5") |
| self.assertEqual(update_check_tag.attrib["_kernel_version_1"], "1.6") |
| self.assertEqual(update_check_tag.attrib["_firmware_version_2"], "1.7") |
| self.assertEqual(update_check_tag.attrib["_kernel_version_2"], "1.8") |
| self.assertEqual(update_check_tag.attrib["_firmware_version_3"], "2.3") |
| self.assertEqual(update_check_tag.attrib["_kernel_version_3"], "3.4") |
| self.assertEqual(update_check_tag.attrib["_firmware_version_4"], "4.3") |
| self.assertEqual(update_check_tag.attrib["_kernel_version_4"], "5.4") |
| |
| def testFailuresPerUrl(self): |
| """Tests response for number of failures allowed per URL.""" |
| self.GenerateAppData() |
| neb = nebraska.Nebraska() |
| neb.UpdateConfig( |
| update_app_index=nebraska.AppIndex(self.tempdir), |
| update_payloads_address=_PAYLOADS_ADDRESS, |
| ) |
| request = GenerateXMLRequest([GenerateXMLAppRequest()]) |
| |
| neb.UpdateConfig(failures_per_url=10) |
| response = neb.GetResponseToRequest(nebraska.Request(request)) |
| root = ElementTree.fromstring(response) |
| action_tag = root.findall("app/updatecheck/manifest/actions/action")[1] |
| self.assertEqual(action_tag.attrib["MaxFailureCountPerUrl"], "10") |
| |
| def testDisablePayloadBackoff(self): |
| """Tests disabling payload backoff on the client.""" |
| self.GenerateAppData() |
| neb = nebraska.Nebraska() |
| neb.UpdateConfig( |
| update_app_index=nebraska.AppIndex(self.tempdir), |
| update_payloads_address=_PAYLOADS_ADDRESS, |
| ) |
| request = GenerateXMLRequest([GenerateXMLAppRequest()]) |
| |
| neb.UpdateConfig(disable_payload_backoff=True) |
| response = neb.GetResponseToRequest(nebraska.Request(request)) |
| root = ElementTree.fromstring(response) |
| action_tag = root.findall("app/updatecheck/manifest/actions/action")[1] |
| self.assertEqual(action_tag.attrib["DisablePayloadBackoff"], "true") |
| |
| def testNumUrls(self): |
| """Tests the number of URLs are passed correctly.""" |
| self.GenerateAppData() |
| neb = nebraska.Nebraska() |
| neb.UpdateConfig( |
| update_app_index=nebraska.AppIndex(self.tempdir), |
| update_payloads_address=_PAYLOADS_ADDRESS, |
| ) |
| request = GenerateXMLRequest([GenerateXMLAppRequest()]) |
| |
| neb.UpdateConfig(num_urls=2) |
| response = neb.GetResponseToRequest(nebraska.Request(request)) |
| root = ElementTree.fromstring(response) |
| urls = root.findall("app/updatecheck/urls/url") |
| self.assertEqual(len(urls), 2) |
| |
| def testEolDate(self): |
| """Tests the EOL date are passed correctly.""" |
| self.GenerateAppData() |
| neb = nebraska.Nebraska() |
| neb.UpdateConfig( |
| update_app_index=nebraska.AppIndex(self.tempdir), |
| update_payloads_address=_PAYLOADS_ADDRESS, |
| ) |
| request = GenerateXMLRequest([GenerateXMLAppRequest()]) |
| |
| neb.UpdateConfig(eol_date=1000) |
| response = neb.GetResponseToRequest(nebraska.Request(request)) |
| root = ElementTree.fromstring(response) |
| update_check = root.find("app/updatecheck") |
| self.assertEqual(update_check.attrib["_eol_date"], "1000") |
| |
| def testExtendedDateAndOptIn(self): |
| """Tests the extended date and opt in are passed correctly.""" |
| self.GenerateAppData() |
| neb = nebraska.Nebraska() |
| neb.UpdateConfig( |
| update_app_index=nebraska.AppIndex(self.tempdir), |
| update_payloads_address=_PAYLOADS_ADDRESS, |
| ) |
| request = GenerateXMLRequest([GenerateXMLAppRequest()]) |
| |
| neb.UpdateConfig(extended_date=20) |
| neb.UpdateConfig(extended_opt_in_required=True) |
| response = neb.GetResponseToRequest(nebraska.Request(request)) |
| root = ElementTree.fromstring(response) |
| update_check = root.find("app/updatecheck") |
| self.assertEqual(update_check.attrib["_extended_date"], "20") |
| self.assertEqual( |
| update_check.attrib["_extended_opt_in_required"], "true" |
| ) |
| |
| def testExtendedOptInFalseIsNotSent(self): |
| """Tests the opt in is not even sent if false.""" |
| self.GenerateAppData() |
| neb = nebraska.Nebraska() |
| neb.UpdateConfig( |
| update_app_index=nebraska.AppIndex(self.tempdir), |
| update_payloads_address=_PAYLOADS_ADDRESS, |
| ) |
| request = GenerateXMLRequest([GenerateXMLAppRequest()]) |
| |
| neb.UpdateConfig(extended_opt_in_required=False) |
| response = neb.GetResponseToRequest(nebraska.Request(request)) |
| root = ElementTree.fromstring(response) |
| update_check = root.find("app/updatecheck") |
| self.assertNotIn("_extended_date", update_check.attrib) |
| self.assertNotIn("_extended_opt_in_required", update_check.attrib) |
| |
| |
| if __name__ == "__main__": |
| # Disable logging so it doesn't pollute the unit test output. Failures and |
| # exceptions are still shown. |
| logging.disable(logging.CRITICAL) |
| |
| unittest.main() |