Browse Source

Add tool to list test cases

Change-Id: Ia7547114c709ac765d51275c9f79fdbb47de8198
changes/41/817241/8
Federico Ressi 7 months ago
parent
commit
debe895033
  1. 25
      tobiko/run/__init__.py
  2. 47
      tobiko/run/_config.py
  3. 165
      tobiko/run/_discover.py
  4. 90
      tobiko/run/_find.py
  5. 53
      tobiko/run/_worker.py
  6. 0
      tobiko/tests/functional/run/__init__.py
  7. 54
      tobiko/tests/functional/run/test_discover.py
  8. 45
      tobiko/tests/functional/run/test_find.py

25
tobiko/run/__init__.py

@ -0,0 +1,25 @@
# Copyright (c) 2021 Red Hat, Inc.
#
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from __future__ import absolute_import
from tobiko.run import _discover
from tobiko.run import _find
discover_test_ids = _discover.discover_test_ids
find_test_ids = _discover.find_test_ids
forked_discover_test_ids = _discover.forked_discover_test_ids
find_test_files = _find.find_test_files

47
tobiko/run/_config.py

@ -0,0 +1,47 @@
# Copyright (c) 2021 Red Hat, Inc.
#
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from __future__ import absolute_import
import typing
import os
import tobiko
class RunConfigFixture(tobiko.SharedFixture):
test_path: typing.List[str]
test_filename: str = 'test_*.py'
python_path: typing.Optional[typing.List[str]] = None
workers_count: typing.Optional[int] = None
def setup_fixture(self):
package_file = os.path.realpath(os.path.realpath(tobiko.__file__))
package_dir = os.path.dirname(package_file)
tobiko_dir = os.path.dirname(package_dir)
self.test_path = [os.path.join(tobiko_dir, 'tobiko', 'tests')]
@property
def forked(self) -> bool:
return self.workers_count != 1
def run_confing(obj=None) -> RunConfigFixture:
if obj is None:
return tobiko.setup_fixture(RunConfigFixture)
fixture = tobiko.get_fixture(obj)
tobiko.check_valid_type(fixture, RunConfigFixture)
return tobiko.setup_fixture(fixture)

165
tobiko/run/_discover.py

@ -0,0 +1,165 @@
# Copyright (c) 2021 Red Hat, Inc.
#
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from __future__ import absolute_import
import inspect
import os
import sys
import typing
import unittest
from oslo_log import log
import tobiko
from tobiko.run import _config
from tobiko.run import _find
from tobiko.run import _worker
LOG = log.getLogger(__name__)
def find_test_ids(test_path: typing.Iterable[str],
test_filename: str = None,
python_path: typing.Iterable[str] = None,
forked: bool = None,
config: _config.RunConfigFixture = None) \
-> typing.List[str]:
config = _config.run_confing(config)
test_files = _find.find_test_files(test_path=test_path,
test_filename=test_filename,
config=config)
if not python_path:
python_path = config.python_path
if forked is None:
forked = bool(config.forked)
if forked:
return forked_discover_test_ids(test_files=test_files,
python_path=python_path)
else:
return discover_test_ids(test_files=test_files,
python_path=python_path)
def discover_test_ids(test_files: typing.Iterable[str],
python_path: typing.Iterable[str] = None) \
-> typing.List[str]:
if not python_path:
python_path = sys.path
python_dirs = [os.path.realpath(p) + '/'
for p in python_path
if os.path.isdir(p)]
test_ids: typing.List[str] = []
for test_file in test_files:
test_ids.extend(discover_file_test_ids(test_file=test_file,
python_dirs=python_dirs))
return test_ids
def discover_file_test_ids(test_file: str,
python_dirs: typing.Iterable[str]) \
-> typing.List[str]:
test_file = os.path.realpath(test_file)
if not os.path.isfile(test_file):
raise ValueError(f"Test file doesn't exist: '{test_file}'")
if not test_file.endswith('.py'):
raise ValueError(f"Test file hasn't .py suffix: '{test_file}'")
for python_dir in python_dirs:
if test_file.startswith(python_dir):
module_name = test_file[len(python_dir):-3].replace('/', '.')
return discover_module_test_ids(module_name)
raise ValueError(f"Test file not in Python path: '{test_file}'")
def discover_module_test_ids(module_name: str) -> typing.List[str]:
LOG.debug(f"Load test module '{module_name}'...")
module = tobiko.load_module(module_name)
test_file = module.__file__
LOG.debug("Inspect test module:\n"
f" module: '{module_name}'\n"
f" filename: '{test_file}'\n")
test_ids: typing.List[str] = []
for obj_name in dir(module):
try:
obj = getattr(module, obj_name)
except AttributeError:
LOG.warning("Error getting object "
f"'{module_name}.{obj_name}'",
exc_info=1)
continue
if (inspect.isclass(obj) and
issubclass(obj, unittest.TestCase) and
not inspect.isabstract(obj)):
LOG.debug("Inspect test class members...\n"
f" file: '{test_file}'\n"
f" module: '{module_name}'\n"
f" object: '{obj_name}'\n")
for member_name in dir(obj):
if member_name.startswith('test_'):
member_id = f"{module_name}.{obj_name}.{member_name}"
try:
member = getattr(obj, member_name)
except Exception:
LOG.error(f'Error getting "{member_id}"', exc_info=1)
continue
if not callable(member):
LOG.error("Class member is not callable: "
f"'{member_id}'")
continue
test_ids.append(member_id)
return test_ids
def forked_discover_test_ids(test_files: typing.Iterable[str],
python_path: typing.Iterable[str] = None) \
-> typing.List[str]:
results = [_worker.call_async(discover_test_ids,
test_files=[test_file],
python_path=python_path)
for test_file in test_files]
test_ids: typing.List[str] = []
for result in results:
test_ids.extend(result.get())
return test_ids
def main(test_path: typing.Iterable[str] = None,
test_filename: str = None,
forked: bool = None,
python_path: typing.Iterable[str] = None):
if test_path is None:
test_path = sys.argv[1:]
try:
test_ids = find_test_ids(test_path=test_path,
test_filename=test_filename,
forked=forked,
python_path=python_path)
except Exception as ex:
sys.stderr.write(f'{ex}\n')
sys.exit(1)
else:
output = ''.join(f'{test_id}\n'
for test_id in test_ids)
sys.stdout.write(output)
sys.exit(0)
if __name__ == '__main__':
main()

90
tobiko/run/_find.py

@ -0,0 +1,90 @@
# Copyright (c) 2021 Red Hat, Inc.
#
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from __future__ import absolute_import
import os
import subprocess
import sys
import typing
from oslo_log import log
from tobiko.run import _config
LOG = log.getLogger(__name__)
def find_test_files(test_path: typing.Iterable[str] = None,
test_filename: str = None,
config: _config.RunConfigFixture = None) \
-> typing.List[str]:
config = _config.run_confing(config)
if test_path:
test_path = list(test_path)
if not test_path:
test_path = config.test_path
if not test_filename:
test_filename = config.test_filename
test_files: typing.List[str] = []
for path in test_path:
path = os.path.realpath(path)
if os.path.isfile(path):
test_files.append(path)
LOG.debug("Found test file:\n"
f" {path}\n",)
continue
if os.path.isdir(path):
find_dir = path
find_name = test_filename
else:
find_dir = os.path.dirname(path)
find_name = os.path.basename(path)
LOG.debug("Find test files...\n"
f" dir: '{find_dir}'\n"
f" name: '{find_name}'\n")
output = subprocess.check_output(
['find', find_dir, '-name', find_name],
universal_newlines=True)
for line in output.splitlines():
line = line.strip()
if line:
assert os.path.isfile(line)
test_files.append(line)
LOG.debug("Found test file(s):\n"
" %s", ' \n'.join(test_files))
return test_files
def main(test_path: typing.List[str] = None):
if test_path is None:
test_path = sys.argv[1:]
try:
test_files = find_test_files(test_path=test_path)
except Exception as ex:
sys.stderr.write(f'{ex}\n')
sys.exit(1)
else:
output = ''.join(f'{test_file}\n'
for test_file in test_files)
sys.stdout.write(output)
sys.exit(0)
if __name__ == '__main__':
main()

53
tobiko/run/_worker.py

@ -0,0 +1,53 @@
# Copyright (c) 2021 Red Hat, Inc.
#
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from __future__ import absolute_import
import multiprocessing.pool
import typing
import tobiko
from tobiko.run import _config
class WorkersPoolFixture(tobiko.SharedFixture):
config: _config.RunConfigFixture = tobiko.required_fixture(
_config.RunConfigFixture)
pool: multiprocessing.pool.Pool
workers_count: int = 0
def __init__(self, workers_count: int = None):
super().__init__()
if workers_count is not None:
self.workers_count = workers_count
def setup_fixture(self):
workers_count = self.workers_count
if not workers_count:
workers_count = self.config.workers_count
self.workers_count = workers_count or 0
self.pool = multiprocessing.pool.Pool(processes=workers_count or None)
def workers_pool() -> multiprocessing.pool.Pool:
return tobiko.setup_fixture(WorkersPoolFixture).pool
def call_async(func: typing.Callable,
*args,
**kwargs) -> multiprocessing.pool.AsyncResult:
return workers_pool().apply_async(func, args=args, kwds=kwargs)

0
tobiko/tests/functional/run/__init__.py

54
tobiko/tests/functional/run/test_discover.py

@ -0,0 +1,54 @@
# Copyright (c) 2019 Red Hat, Inc.
#
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from __future__ import absolute_import
import os
import typing
import testtools
from tobiko import run
class DiscoverTestIdsTest(testtools.TestCase):
def test_discover_test_ids(self,
test_files: typing.List[str] = None):
if test_files is None:
test_files = [__file__]
test_ids = run.discover_test_ids(test_files=test_files)
self.assertIn(self.id(), test_ids)
def test_forked_discover_test_ids(self,
test_files: typing.List[str] = None):
if test_files is None:
test_files = [__file__]
test_ids = run.forked_discover_test_ids(test_files=test_files)
self.assertIn(self.id(), test_ids)
def test_find_test_ids(self,
test_path: typing.List[str] = None,
forked=False):
if test_path is None:
test_path = [__file__]
test_ids = run.find_test_ids(test_path=test_path, forked=forked)
self.assertIn(self.id(), test_ids)
def test_find_test_ids_with_test_dir(self):
self.test_find_test_ids(test_path=[os.path.dirname(__file__)])
def test_find_test_ids_with_forked(self):
self.test_find_test_ids(forked=True)

45
tobiko/tests/functional/run/test_find.py

@ -0,0 +1,45 @@
# Copyright (c) 2019 Red Hat, Inc.
#
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from __future__ import absolute_import
import os
import typing
import testtools
from tobiko import run
class FindTestFilesTest(testtools.TestCase):
def test_find_test_files(self,
test_path: typing.List[str] = None):
if test_path is None:
test_path = [__file__]
test_files = run.find_test_files(test_path=test_path)
self.assertIn(__file__, test_files)
test_path = [os.path.realpath(path)
for path in test_path]
for test_file in test_files:
for path in test_path:
if test_file.startswith(path):
break
else:
self.fail(f"File '{test_file}' not in any path ({test_path})")
def test_find_test_files_with_test_dir(self):
return self.test_find_test_files(
test_path=[os.path.dirname(__file__)])
Loading…
Cancel
Save