blob: ff263033dc3cdd77919f1cf60c23e62354d4e962 [file] [log] [blame] [edit]
# Copyright 2023 The ChromiumOS Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Unit tests for portage_wrapper.py."""
import os
import shutil
import subprocess
import sys
import tempfile
from typing import List, Tuple
import unittest
_PORTAGEQ_WRAPPER_PATH = (
"bazel/portage/bin/fast_install_packages/portageq_wrapper.py"
)
_FAKE_PORTAGEQ_PATH = (
"bazel/portage/bin/fast_install_packages/testdata/fake_portageq.py"
)
class PortageqTest(unittest.TestCase):
"""Unit tests for portageq_wrapper.py."""
def setUp(self):
self._test_root = tempfile.mkdtemp()
os.makedirs(os.path.join(self._test_root, "build/meow"))
def tearDown(self):
self._verify_no_temporary_files()
shutil.rmtree(self._test_root)
def _verify_no_temporary_files(self):
"""Verifies that there is no temporary files left in the test root."""
for dirpath, _, filenames in os.walk(self._test_root):
for filename in filenames:
self.assertFalse(
filename.startswith("."),
"Temporary file left at %s"
% os.path.join(dirpath, filename),
)
def _run_portageq_wrapper(self, args: List[str]) -> Tuple[str, int]:
"""Runs the portageq wrapper (portageq_wrapper.py) with fakes.
We set up environment variables so that:
- The wrapper will call into the fake portageq (fake_portageq.py)
instead of the system-installed /usr/bin/portageq.
- The wrapper will create cache files under `self._test_root` instead
of the host file system.
Args:
args: Arguments to pass to the portageq wrapper.
Returns:
(stdout, count) where:
stdout: The standard output from the process.
count: The number of times the real portageq (fake_portageq.py) was
executed. We count this by inspecting the standard error.
"""
env = os.environ.copy()
env.update(
{
"PORTAGEQ_WRAPPER_REAL_PORTAGEQ": _FAKE_PORTAGEQ_PATH,
"PORTAGEQ_WRAPPER_CACHE_DIR_PREFIX": self._test_root,
}
)
with subprocess.Popen(
[_PORTAGEQ_WRAPPER_PATH] + args,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
encoding="ascii",
env=env,
) as proc:
stdout, stderr = proc.communicate()
sys.stderr.write(stderr) # Pass-through stderr
if proc.returncode != 0:
raise subprocess.CalledProcessError(proc.returncode, proc.args)
count = stderr.count("[fake_portageq] Command:")
return stdout, count
def test_get_repos_cached(self):
"""Verifies that get_repos is cached."""
self.assertEqual(
self._run_portageq_wrapper(["get_repos", "/build/meow/"]),
("meow chromiumos portage-stable\n", 1),
)
self.assertEqual(
self._run_portageq_wrapper(["get_repos", "/build/meow/"]),
("meow chromiumos portage-stable\n", 0),
)
self.assertEqual(
self._run_portageq_wrapper(["get_repos", "/"]),
("chromiumos portage-stable\n", 1),
)
self.assertEqual(
self._run_portageq_wrapper(["get_repos", "/"]),
("chromiumos portage-stable\n", 0),
)
def test_get_repo_path_cached(self):
"""Verifies that get_repo_path is cached."""
self.assertEqual(
self._run_portageq_wrapper(
[
"get_repo_path",
"/build/meow/",
"portage-stable",
"chromiumos",
]
),
(
"/mnt/host/source/src/third_party/portage-stable\n"
"/mnt/host/source/src/third_party/chromiumos-overlay\n",
1,
),
)
self.assertEqual(
self._run_portageq_wrapper(
["get_repo_path", "/build/meow/", "chromiumos", "meow"]
),
(
"/mnt/host/source/src/third_party/chromiumos-overlay\n"
"/mnt/host/source/src/overlays/overlay-meow\n",
1,
),
)
self.assertEqual(
self._run_portageq_wrapper(
[
"get_repo_path",
"/build/meow/",
"portage-stable",
"chromiumos",
"meow",
]
),
(
"/mnt/host/source/src/third_party/portage-stable\n"
"/mnt/host/source/src/third_party/chromiumos-overlay\n"
"/mnt/host/source/src/overlays/overlay-meow\n",
0,
),
)
self.assertEqual(
self._run_portageq_wrapper(
["get_repo_path", "/", "portage-stable", "chromiumos"]
),
(
"/mnt/host/source/src/third_party/portage-stable\n"
"/mnt/host/source/src/third_party/chromiumos-overlay\n",
1,
),
)
self.assertEqual(
self._run_portageq_wrapper(["get_repo_path", "/", "chromiumos"]),
("/mnt/host/source/src/third_party/chromiumos-overlay\n", 0),
)
def test_get_repo_path_unknown_repos(self):
"""Verifies that get_repo_path fails for unknown repositories."""
with self.assertRaises(subprocess.CalledProcessError):
self._run_portageq_wrapper(
["get_repo_path", "/build/meow/", "portage-stable", "eve"]
)
with self.assertRaises(subprocess.CalledProcessError):
self._run_portageq_wrapper(
["get_repo_path", "/", "portage-stable", "meow"]
)
def test_get_repos_invalid_root(self):
"""Verifies that get_repos fails for invalid root directories."""
with self.assertRaises(subprocess.CalledProcessError):
self._run_portageq_wrapper(["get_repos", "/build/eve/"])
def test_get_repo_path_invalid_root(self):
"""Verifies that get_repo_path fails for invalid root directories."""
with self.assertRaises(subprocess.CalledProcessError):
self._run_portageq_wrapper(
["get_repo_path", "/build/eve/", "portage-stable"]
)
if __name__ == "__main__":
unittest.main()