Issue 6602 - Introduce watchextensions

tests/test_ExtWatch.py

+# This Source Code Form is subject to the terms of the Mozilla Public
+# License, v. 2.0. If a copy of the MPL was not distributed with this
+# file, You can obtain one at http://mozilla.org/MPL/2.0/.
+
+import io
+import os
+import pytest
+import shutil
+import subprocess
+from zipfile import ZipFile
+try:  # Python 3+
+    from unittest import mock
+except ImportError:  # Python 2.7
+    import mock
+
+from watchextensions import (OMAHA_URL, ConfigurationError, ExtWatcher,
+                             read_config)
+
+
+@pytest.fixture
+def remote_local(tmpdir):
+    remote = tmpdir.mkdir('remote')
+    subprocess.check_output(['git', 'init', '--bare'], cwd=str(remote))
+
+    local = tmpdir.mkdir('repository')
+    subprocess.check_output(['git', 'clone', str(remote), str(local)])
+
+    local.join('a.txt').write('foo\nbar')
+
+    subprocess.check_output(['git', 'add', '--all'], cwd=str(local))
+    subprocess.check_output(['git', 'commit', '-m', '0.1'], cwd=str(local))
+    subprocess.check_output(['git', 'push', 'origin', 'master'],
+                            cwd=str(local))
+
+    return remote
+
+
+@pytest.fixture
+def test_config(tmpdir, remote_local):
+    config = tmpdir.join('test.ini')
+    config.write("""
+    [extensions]
+    test_id=abc
+    test_repository={}
+    [enabled]
+    test=
+    """.format(str(remote_local)))
+    return str(config)
+
+
+@pytest.fixture
+def errorneous_configs(tmpdir):
+    folder = tmpdir.mkdir('configs')
+    no_section = folder.join('no_section.ini')
+    no_section.write('[enabled]\ntest=')
+    no_option = folder.join('no_option.ini')
+    no_option.write('[extensions]\na_id=a\n[enabled]\na=')
+    empty_option = folder.join('empty_option.ini')
+    empty_option.write('[extensions]\na_id=a\na_repository=\n[enabled]\na=')
+
+    return str(no_section), str(no_option), str(empty_option)
+
+
+@pytest.fixture
+def crx(tmpdir):
+    v2_path = str(tmpdir.join('v2.crx'))
+
+    with ZipFile(v2_path, 'w') as fp:
+        fp.writestr('a.txt', 'foo\nbaz')
+
+    return str(v2_path)
+
+
+def mock_urlopen(*args, **kwargs):
+    mocked_xml = ('<gupdate xmlns="http://www.google.com/update2/response" '
+                  'protocol="2.0" server="prod"><app appid="abc" cohort="" '
+                  'cohortname="" status="ok"><updatecheck version="0.2"/>'
+                  '</app></gupdate>').encode()
+    if OMAHA_URL in args:
+        fp = io.open(os.path.join(os.path.dirname(__file__),
+                                  'data',
+                                  'omaha.json'),
+                     'rt')
+        mocked = fp.read().encode()
+        fp.close()
+    else:
+        mocked = mocked_xml
+
+    buf = io.BytesIO()
+    buf.write(mocked)
+    buf.seek(0)
+    return buf
+
+
+def test_download_push_and_diff(crx, test_config, tmpdir, remote_local):
+    def mock_urlretrieve(url, filename, *args, **kwargs):
+        shutil.copyfile(crx, filename)
+        return None
+
+    config = read_config(test_config)
+    with ExtWatcher('test', config, True, True) as watcher:
+        with mock.patch('watchextensions.urllib.urlopen',
+                        side_effect=mock_urlopen):
+            with mock.patch('watchextensions.urllib.urlretrieve',
+                            side_effect=mock_urlretrieve):
+                watcher.run()
+
+    diff = subprocess.check_output(['git', 'show'], cwd=str(remote_local))
+    assert '-bar'.encode() in diff
+    assert '+baz'.encode() in diff
+
+
+def test_config_errors(tmpdir, errorneous_configs, caplog):
+    no_section, no_option, empty_option = errorneous_configs
+
+    with pytest.raises(ConfigurationError):
+        read_config(os.path.join(str(tmpdir), 'nonextistent'))
+
+    for cfg, msg in [(no_section, 'Section [extensions] is missing!'),
+                     (no_option, '"a" is not fully configured!'),
+                     (empty_option, '"a" is not fully configured!')]:
+
+        config = read_config(cfg)
+        with ExtWatcher('a', config, False, False) as watcher:
+            watcher.run()
+
+        records = caplog.get_records('call')
+
+        assert msg in records[-1].message