Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/svn/tests/test_utils.py
| # Copyright (C) 2016-2021 The Software Heritage developers | # Copyright (C) 2016-2021 The Software Heritage developers | ||||
| # See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
| # License: GNU General Public License version 3, or any later version | # License: GNU General Public License version 3, or any later version | ||||
| # See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
| import logging | import logging | ||||
| import os | import os | ||||
| import pty | import pty | ||||
| import shutil | import shutil | ||||
| from subprocess import Popen | from subprocess import Popen | ||||
| from swh.loader.svn import utils | from swh.loader.svn import utils | ||||
| from swh.model.model import Timestamp | |||||
| def test_outputstream(): | def test_outputstream(): | ||||
| stdout_r, stdout_w = pty.openpty() | stdout_r, stdout_w = pty.openpty() | ||||
| echo = Popen(["echo", "-e", "foo\nbar\nbaz"], stdout=stdout_w) | echo = Popen(["echo", "-e", "foo\nbar\nbaz"], stdout=stdout_w) | ||||
| os.close(stdout_w) | os.close(stdout_w) | ||||
| stdout_stream = utils.OutputStream(stdout_r) | stdout_stream = utils.OutputStream(stdout_r) | ||||
| lines = [] | lines = [] | ||||
| while True: | while True: | ||||
| current_lines, readable = stdout_stream.read_lines() | current_lines, readable = stdout_stream.read_lines() | ||||
| lines += current_lines | lines += current_lines | ||||
| if not readable: | if not readable: | ||||
| break | break | ||||
| echo.wait() | echo.wait() | ||||
| os.close(stdout_r) | os.close(stdout_r) | ||||
| assert lines == ["foo", "bar", "baz"] | assert lines == ["foo", "bar", "baz"] | ||||
| def test_strdate_to_timestamp(): | |||||
| """Formatted string date should be converted in timestamp.""" | |||||
| actual_ts = utils.strdate_to_timestamp("2011-05-31T06:04:39.800722Z") | |||||
| assert actual_ts == Timestamp(seconds=1306821879, microseconds=800722) | |||||
| actual_ts = utils.strdate_to_timestamp("2011-05-31T06:03:39.123450Z") | |||||
| assert actual_ts == Timestamp(seconds=1306821819, microseconds=123450) | |||||
| def test_strdate_to_timestamp_empty_does_not_break(): | |||||
| """Empty or None date should be timestamp 0.""" | |||||
| default_ts = Timestamp(seconds=0, microseconds=0) | |||||
| assert default_ts == utils.strdate_to_timestamp("") | |||||
| assert default_ts == utils.strdate_to_timestamp(None) | |||||
| def test_init_svn_repo_from_dump(datadir, tmp_path): | def test_init_svn_repo_from_dump(datadir, tmp_path): | ||||
| """Mounting svn repository out of a dump is ok""" | """Mounting svn repository out of a dump is ok""" | ||||
| dump_name = "penguinsdbtools2018.dump.gz" | dump_name = "penguinsdbtools2018.dump.gz" | ||||
| dump_path = os.path.join(datadir, dump_name) | dump_path = os.path.join(datadir, dump_name) | ||||
| tmp_repo, repo_path = utils.init_svn_repo_from_dump( | tmp_repo, repo_path = utils.init_svn_repo_from_dump( | ||||
| dump_path, gzip=True, cleanup_dump=False, root_dir=tmp_path | dump_path, gzip=True, cleanup_dump=False, root_dir=tmp_path | ||||
| ) | ) | ||||
| ▲ Show 20 Lines • Show All 86 Lines • Show Last 20 Lines | |||||