super().__init__(name, test_manager)
def populate_tests(self, uri_minfo_special_scenarios, scenarios):
+ validatetests = []
for root, _, files in os.walk(self.tests_dir):
for f in files:
name, ext = os.path.splitext(f)
if ext != ".validatetest":
continue
- fpath = os.path.abspath(os.path.join(root, f))
- pathname = os.path.abspath(os.path.join(root, name))
- name = pathname.replace(os.path.commonpath([self.tests_dir, root]), '').replace('/', '.')
- self.add_test(GstValidateSimpleTest(fpath, 'test' + name,
- self.test_manager.options,
- self.test_manager.reporter))
+ validatetests.append(os.path.abspath(os.path.join(root, f)))
+
+ for validatetest in self.test_manager.scenarios_manager.discover_scenarios(validatetests):
+ pathname, _ext = os.path.splitext(validatetest.path)
+ name = pathname.replace(os.path.commonpath([self.tests_dir, root]), '').replace('/', '.')
+
+ self.add_test(GstValidateSimpleTest(validatetest.path,
+ 'test' + name,
+ self.test_manager.options,
+ self.test_manager.reporter,
+ test_info=validatetest))
class GstValidatePipelineTestsGenerator(GstValidateTestsGenerator):
class GstValidateSimpleTest(GstValidateTest):
- def __init__(self, test_file, *args, **kwargs):
+ def __init__(self, test_file, *args, test_info=None, **kwargs):
self.test_file = test_file
+ self.test_info = test_info
+
super().__init__(GstValidateBaseTestManager.COMMAND, *args, **kwargs)
def build_arguments(self):
if self.options.mute:
self.add_arguments('--use-fakesinks')
+ def needs_http_server(self):
+ try:
+ return bool(self.test_info.needs_http_server)
+ except AttributeError:
+ return False
+
class GstValidateLaunchTest(GstValidateTest):
def needs_http_server(self):
for test in self.list_tests():
- if self._is_test_wanted(test) and test.media_descriptor is not None:
- protocol = test.media_descriptor.get_protocol()
- uri = test.media_descriptor.get_uri()
- uri_requires_http_server = False
- if uri:
- if 'http-server-port' in uri:
- expanded_uri = uri % {
- 'http-server-port': self.options.http_server_port}
- uri_requires_http_server = expanded_uri.find(
- "127.0.0.1:%s" % self.options.http_server_port) != -1
- if protocol in [Protocols.HTTP, Protocols.HLS, Protocols.DASH] or uri_requires_http_server:
+ if self._is_test_wanted(test):
+ if test.needs_http_server():
return True
+
return False
def set_settings(self, options, args, reporter):
else:
self.scenario = scenario
+ def needs_http_server(self):
+ if self.media_descriptor is None:
+ return False
+
+ protocol = self.media_descriptor.get_protocol()
+ uri = self.media_descriptor.get_uri()
+ uri_requires_http_server = False
+ if uri:
+ if 'http-server-port' in uri:
+ expanded_uri = uri % {
+ 'http-server-port': self.options.http_server_port}
+ uri_requires_http_server = expanded_uri.find(
+ "127.0.0.1:%s" % self.options.http_server_port) != -1
+ if protocol in [Protocols.HTTP, Protocols.HLS, Protocols.DASH] or uri_requires_http_server:
+ return True
+
+ return False
+
def kill_subprocess(self):
Test.kill_subprocess(self)