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 ConfigurationError, ExtWatcher, read_config, urllib
+
+
+@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)
+
+
+orig_urlopen = urllib.urlopen
+
+
+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 'https://omahaproxy.appspot.com/all.json?os=win' in args:
+        return orig_urlopen(*args)

[Vasily Kuznetsov, 2018/05/15 16:49:31]

This makes the test hit that API every time it's run and makes it dependent on
the internet in general. It would be better to mock it, unless it's too much
work.

[tlucas, 2018/05/16 09:52:37]

No, you are right. I'm also mocking this now.

+    buf = io.BytesIO()
+    buf.write(mocked_xml)
+    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