Commit 6a076192 authored by Alexandru Băluț's avatar Alexandru Băluț

tests: Support running the tests with common runners

GES caches assets and this leads to tests failing. To make sure the
cache is not used we copy the assets and use the copies instead.

Fixes https://phabricator.freedesktop.org/T7920
parent f0face5f
......@@ -22,10 +22,12 @@ A collection of objects to use for testing
import contextlib
import gc
import os
import shutil
import tempfile
import unittest
from unittest import mock
from gi.repository import GES
from gi.repository import GLib
from gi.repository import Gst
from gi.repository import Gtk
......@@ -36,12 +38,13 @@ from pitivi.project import ProjectManager
from pitivi.settings import GlobalSettings
from pitivi.timeline.timeline import TimelineContainer
from pitivi.utils.loggable import Loggable
from pitivi.utils.misc import path_from_uri
from pitivi.utils.proxy import ProxyingStrategy
from pitivi.utils.proxy import ProxyManager
from pitivi.utils.timeline import Selected
detect_leaks = os.environ.get("PITIVI_TEST_DETECT_LEAKS", "0") not in ("0", "")
os.environ["PITIVI_USER_CACHE_DIR"] = tempfile.mkdtemp("pitiviTestsuite")
os.environ["PITIVI_USER_CACHE_DIR"] = tempfile.mkdtemp(suffix="pitiviTestsuite")
def clean_pitivi_mock(app):
......@@ -216,9 +219,8 @@ class TestCase(unittest.TestCase, Loggable):
@contextlib.contextmanager
def created_project_file(asset_uri="file:///icantpossiblyexist.png"):
"""
Create a project file.
def created_project_file(asset_uri):
"""Creates a project file.
Yields:
str: The URI of the new project
......@@ -247,9 +249,29 @@ def created_project_file(asset_uri="file:///icantpossiblyexist.png"):
os.remove(xges_path)
def get_sample_uri(sample):
tests_dir = os.path.dirname(os.path.abspath(__file__))
return Gst.filename_to_uri(os.path.join(tests_dir, "samples", sample))
def get_sample_uri(sample, tests_dir=None):
if not tests_dir:
tests_dir = os.path.dirname(os.path.abspath(__file__))
tests_dir = os.path.join(tests_dir, "samples")
return Gst.filename_to_uri(os.path.join(tests_dir, sample))
@contextlib.contextmanager
def cloned_sample(*samples):
"""Gets a context manager which commits the transaction at the end."""
tmpdir = tempfile.mkdtemp(suffix="pitivi_cloned_samples")
module = globals()
original_get_sample_uri = module["get_sample_uri"]
module["get_sample_uri"] = lambda sample: original_get_sample_uri(sample, tests_dir=tmpdir)
for sample in samples:
sample_path = path_from_uri(original_get_sample_uri(sample))
clone_path = path_from_uri(get_sample_uri(sample))
shutil.copyfile(sample_path, clone_path)
try:
yield tmpdir
finally:
module["get_sample_uri"] = original_get_sample_uri
shutil.rmtree(tmpdir)
def get_clip_children(ges_clip, *track_types, recursive=False):
......@@ -258,16 +280,6 @@ def get_clip_children(ges_clip, *track_types, recursive=False):
yield ges_timeline_element
def clean_proxy_samples():
_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "samples")
proxy_manager = ProxyManager(mock.MagicMock())
for f in os.listdir(_dir):
if f.endswith(proxy_manager.proxy_extension):
f = os.path.join(_dir, f)
os.remove(f)
def create_test_clip(clip_type):
clip = clip_type()
clip.selected = Selected()
......
[junit-xml]
always-on = True
path = @path@
......@@ -43,8 +43,6 @@ class PitiviTest(Test):
def setup_tests(test_manager, options):
"""Sets up Pitivi unit testsuite."""
print("Forcing 1 job at a time as testsuite will fail otherwise")
options.num_jobs = 1
loader = unittest.TestLoader()
testsuites = loader.discover(CDIR)
for testsuite in testsuites:
......
......@@ -17,7 +17,7 @@
# Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
# Boston, MA 02110-1301, USA.
"""
Test for pitivi/mainwindow.py
Tests for pitivi/mainwindow.py
"""
from unittest import mock
......@@ -31,14 +31,10 @@ from tests import common
class TestMainWindow(common.TestCase):
"""
Test MainWindow
"""
def testSwitchContextTab(self):
"""
Test tab switches
"""
"""Tests for the MainWindow class."""
def test_switch_context_tab(self):
"""Checks tab switches."""
app = common.create_pitivi_mock()
mainwindow = MainWindow(app)
for expected_tab, b_element in [
......@@ -46,13 +42,13 @@ class TestMainWindow(common.TestCase):
(0, GES.SourceClip()),
(1, GES.TransitionClip())]:
mainwindow.switchContextTab(b_element)
self.assertEqual(expected_tab,
mainwindow.context_tabs.get_current_page(),
self.assertEqual(mainwindow.context_tabs.get_current_page(),
expected_tab,
b_element)
# Make sure the tab does not change when using an invalid argument.
mainwindow.switchContextTab("invalid")
self.assertEqual(
expected_tab, mainwindow.context_tabs.get_current_page())
self.assertEqual(mainwindow.context_tabs.get_current_page(),
expected_tab)
mainwindow.destroy()
......@@ -65,8 +61,7 @@ class TestMainWindow(common.TestCase):
mainwindow = MainWindow(app)
mainwindow.viewer = mock.MagicMock()
def __pm_missing_uri_cb(project_manager, project,
error, asset):
def __pm_missing_uri_cb(project_manager, project, error, asset):
nonlocal mainloop
nonlocal mainwindow
nonlocal self
......@@ -109,22 +104,19 @@ class TestMainWindow(common.TestCase):
disconnectAllByFunc(app.project_manager,
mainwindow._projectManagerNewProjectFailedCb)
app.project_manager.connect("missing-uri",
__pm_missing_uri_cb)
app.project_manager.connect("missing-uri", __pm_missing_uri_cb)
with common.created_project_file() as uri:
app.project_manager.loadProject(uri)
with common.cloned_sample():
asset_uri = common.get_sample_uri("missing.png")
with common.created_project_file(asset_uri) as uri:
app.project_manager.loadProject(uri)
mainloop.run()
def test_loading_project_no_proxy(self):
"""
Test loading failure without proxies
"""
"""Checks loading failure without proxies."""
self.__loading_failure(has_proxy=False)
def test_loading_project_with_proxy(self):
"""
Test loading failure with proxies
"""
"""Checks loading failure with proxies."""
self.__loading_failure(has_proxy=True)
......@@ -98,8 +98,8 @@ class BaseTestMediaLibrary(common.TestCase):
self.app.project_manager.current_project.create_asset(
common.get_sample_uri(sample_name), GES.UriClip)
def runCheckImport(self, assets, proxying_strategy=ProxyingStrategy.ALL,
check_no_transcoding=False, clean_proxies=True):
def check_import(self, assets, proxying_strategy=ProxyingStrategy.ALL,
check_no_transcoding=False):
self._customSetUp(proxyingStrategy=proxying_strategy,
numTranscodingJobs=4,
lastClipView=medialibrary.SHOW_TREEVIEW)
......@@ -108,9 +108,6 @@ class BaseTestMediaLibrary(common.TestCase):
self.medialibrary._progressbar.connect(
"notify::fraction", self._progressBarCb)
if clean_proxies:
common.clean_proxy_samples()
self._createAssets(assets)
self.mainloop.run()
self.assertFalse(self.medialibrary._progressbar.props.visible)
......@@ -120,7 +117,7 @@ class TestMediaLibrary(BaseTestMediaLibrary):
def stop_using_proxies(self, delete_proxies=False):
sample_name = "30fps_numeroted_frames_red.mkv"
self.runCheckImport([sample_name])
self.check_import([sample_name])
asset_uri = common.get_sample_uri(sample_name)
proxy = self.medialibrary.storemodel[0][medialibrary.COL_ASSET]
......@@ -137,90 +134,102 @@ class TestMediaLibrary(BaseTestMediaLibrary):
self.assertEqual(self.medialibrary.storemodel[0][medialibrary.COL_URI],
asset_uri)
def testTranscoding(self):
self.runCheckImport(["30fps_numeroted_frames_red.mkv"])
def test_transcoding_and_reusing(self):
sample_name = "30fps_numeroted_frames_red.mkv"
with common.cloned_sample(sample_name):
# Create proxies.
self.check_import([sample_name])
# Try to import again, checking that no transcoding is done.
self.check_import([sample_name],
check_no_transcoding=True)
def testDisableProxies(self):
self.runCheckImport(["30fps_numeroted_frames_red.mkv"],
ProxyingStrategy.NOTHING, True)
def testReuseProxies(self):
# Create proxies
self.runCheckImport(["30fps_numeroted_frames_red.mkv"])
self.info("Now trying to import again, checking that no"
" transcoding is done.")
self.runCheckImport(["30fps_numeroted_frames_red.mkv"],
check_no_transcoding=True,
clean_proxies=False)
sample_name = "30fps_numeroted_frames_red.mkv"
with common.cloned_sample(sample_name):
self.check_import([sample_name],
proxying_strategy=ProxyingStrategy.NOTHING,
check_no_transcoding=True)
def testSaveProjectWithRemovedProxy(self):
sample_name = "30fps_numeroted_frames_red.mkv"
self.runCheckImport([sample_name])
with common.cloned_sample(sample_name):
self.check_import([sample_name])
project = self.app.project_manager.current_project
asset = GES.UriClipAsset.request_sync(common.get_sample_uri(sample_name))
target = asset.get_proxy_target()
self.assertEqual(set(project.list_assets(GES.Extractable)), set([target, asset]))
project = self.app.project_manager.current_project
asset = GES.UriClipAsset.request_sync(common.get_sample_uri(sample_name))
target = asset.get_proxy_target()
self.assertEqual(set(project.list_assets(GES.Extractable)), set([target, asset]))
# Remove the asset
self.medialibrary.remove_assets_action.emit("activate", None)
# Remove the asset
self.medialibrary.remove_assets_action.emit("activate", None)
# Make sure that the project has not assets anymore
self.assertEqual(project.list_assets(GES.Extractable), [])
# Make sure that the project has not assets anymore
self.assertEqual(project.list_assets(GES.Extractable), [])
# Save the project and reload it, making sure there is no asset
# in that new project
project_uri = Gst.filename_to_uri(tempfile.NamedTemporaryFile().name)
project.save(project.ges_timeline, project_uri, None, True)
# Save the project and reload it, making sure there is no asset
# in that new project
project_uri = Gst.filename_to_uri(tempfile.NamedTemporaryFile().name)
project.save(project.ges_timeline, project_uri, None, True)
self._customSetUp(project_uri)
self.assertNotEqual(project, self.app.project_manager.current_project)
project = self.app.project_manager.current_project
self.assertEqual(project.list_assets(GES.Extractable), [])
self._customSetUp(project_uri)
self.assertNotEqual(project, self.app.project_manager.current_project)
project = self.app.project_manager.current_project
self.assertEqual(project.list_assets(GES.Extractable), [])
def testNewlyImportedAssetSelected(self):
self.runCheckImport(["30fps_numeroted_frames_red.mkv",
"30fps_numeroted_frames_blue.webm"])
samples = ["30fps_numeroted_frames_red.mkv",
"30fps_numeroted_frames_blue.webm"]
with common.cloned_sample(*samples):
self.check_import(samples)
self.assertEqual(len(list(self.medialibrary.getSelectedPaths())),
len(self.samples))
def test_stop_using_proxies(self):
self.stop_using_proxies()
sample_name = "30fps_numeroted_frames_red.mkv"
with common.cloned_sample(sample_name):
self.stop_using_proxies()
def test_delete_proxy(self):
self.stop_using_proxies(delete_proxies=True)
sample_name = "30fps_numeroted_frames_red.mkv"
with common.cloned_sample(sample_name):
self.stop_using_proxies(delete_proxies=True)
asset = self.medialibrary.storemodel[0][medialibrary.COL_ASSET]
proxy_uri = self.app.proxy_manager.getProxyUri(asset)
asset = self.medialibrary.storemodel[0][medialibrary.COL_ASSET]
proxy_uri = self.app.proxy_manager.getProxyUri(asset)
# Requesting UriClip sync will return None if the asset is not in cache
# this way we make sure that this asset used to exist
proxy = GES.Asset.request(GES.UriClip, proxy_uri)
self.assertIsNotNone(proxy)
self.assertFalse(os.path.exists(Gst.uri_get_location(proxy_uri)))
# Requesting UriClip sync will return None if the asset is not in cache
# this way we make sure that this asset used to exist
proxy = GES.Asset.request(GES.UriClip, proxy_uri)
self.assertIsNotNone(proxy)
self.assertFalse(os.path.exists(Gst.uri_get_location(proxy_uri)))
self.assertIsNone(asset.get_proxy())
self.assertIsNone(asset.get_proxy())
# And let's recreate the proxy file.
self.app.project_manager.current_project.use_proxies_for_assets([asset])
self.assertEqual(asset.creation_progress, 0)
# And let's recreate the proxy file.
self.app.project_manager.current_project.use_proxies_for_assets([asset])
self.assertEqual(asset.creation_progress, 0)
# Check that the info column notifies the user about progress
self.assertTrue("Proxy creation progress:" in
self.medialibrary.storemodel[0][medialibrary.COL_INFOTEXT])
# Check that the info column notifies the user about progress
self.assertTrue("Proxy creation progress:" in
self.medialibrary.storemodel[0][medialibrary.COL_INFOTEXT])
# Run the mainloop and let _progressBarCb stop it when the proxy is
# ready
self.mainloop.run()
# Run the mainloop and let _progressBarCb stop it when the proxy is
# ready
self.mainloop.run()
self.assertEqual(asset.creation_progress, 100)
self.assertEqual(asset.get_proxy(), proxy)
self.assertEqual(asset.creation_progress, 100)
self.assertEqual(asset.get_proxy(), proxy)
def test_supported_out_of_container_audio(self):
self.runCheckImport(["mp3_sample.mp3"], check_no_transcoding=True)
def testMissingUriDisplayed(self):
with common.created_project_file() as uri:
self._customSetUp(project_uri=uri)
sample = "mp3_sample.mp3"
with common.cloned_sample(sample):
self.check_import([sample], check_no_transcoding=True)
def test_missing_uri_displayed(self):
with common.cloned_sample():
asset_uri = common.get_sample_uri("missing.png")
with common.created_project_file(asset_uri) as uri:
self._customSetUp(project_uri=uri)
self.assertTrue(self.medialibrary._import_warning_infobar.props.visible)
......@@ -84,10 +84,12 @@ class TestAudioPreviewer(BaseTestMediaLibrary):
def test_waveform_creation(self):
"""Checks the waveform generation."""
sample_name = "1sec_simpsons_trailer.mp4"
self.runCheckImport([sample_name])
with common.cloned_sample(sample_name):
self.check_import([sample_name])
sample_uri = common.get_sample_uri(sample_name)
wavefile = get_wavefile_location_for_uri(sample_uri)
sample_uri = common.get_sample_uri(sample_name)
wavefile = get_wavefile_location_for_uri(sample_uri)
self.assertTrue(os.path.exists(wavefile), wavefile)
with open(wavefile, "rb") as fsamples:
......
......@@ -126,9 +126,11 @@ class TestProjectManager(TestCase):
result = [False]
self.manager.connect("missing-uri", missingUriCb, result)
with common.created_project_file() as uri:
self.assertTrue(self.manager.loadProject(uri))
mainloop.run()
with common.cloned_sample():
asset_uri = common.get_sample_uri("missing.png")
with common.created_project_file(asset_uri) as uri:
self.assertTrue(self.manager.loadProject(uri))
mainloop.run()
self.assertTrue(result[0], "missing-uri has not been emitted")
def testLoaded(self):
......@@ -139,10 +141,11 @@ class TestProjectManager(TestCase):
self.manager.connect("new-project-loaded", new_project_loaded_cb)
asset_uri = common.get_sample_uri("flat_colour1_640x480.png")
with common.created_project_file(asset_uri=asset_uri) as uri:
self.assertTrue(self.manager.loadProject(uri))
mainloop.run()
with common.cloned_sample("flat_colour1_640x480.png"):
asset_uri = common.get_sample_uri("flat_colour1_640x480.png")
with common.created_project_file(asset_uri=asset_uri) as uri:
self.assertTrue(self.manager.loadProject(uri))
mainloop.run()
project = self.manager.current_project
self.assertFalse(project.at_least_one_asset_missing)
......@@ -393,9 +396,6 @@ class TestProjectLoading(common.TestCase):
proj_uri = self.create_project_file_from_xges(app, PROJECT_STR)
# Remove proxy
common.clean_proxy_samples()
def closing_project_cb(*args, **kwargs):
# Do not ask whether to save project on closing.
return True
......@@ -412,8 +412,10 @@ class TestProjectLoading(common.TestCase):
def test_load_project_with_missing_proxy(self):
"""Checks loading a project with missing proxies."""
mainloop, app, medialib, proxy_uri = self.load_project_with_missing_proxy()
mainloop.run()
with common.cloned_sample("1sec_simpsons_trailer.mp4"):
mainloop, app, medialib, proxy_uri = self.load_project_with_missing_proxy()
mainloop.run()
self.assertEqual(len(medialib.storemodel), 1)
self.assertEqual(medialib.storemodel[0][medialibrary.COL_ASSET].props.id,
proxy_uri)
......@@ -424,49 +426,51 @@ class TestProjectLoading(common.TestCase):
"""Checks progress tracking of loading project with missing proxies."""
from gi.repository import GstTranscoder
# Disable proxy generation by not making it start ever.
# This way we are sure it will not finish before we test
# the state while it is being rebuilt.
with mock.patch.object(GstTranscoder.Transcoder, "run_async"):
mainloop, app, medialib, proxy_uri = self.load_project_with_missing_proxy()
uri = common.get_sample_uri("1sec_simpsons_trailer.mp4")
with common.cloned_sample("1sec_simpsons_trailer.mp4"):
# Disable proxy generation by not making it start ever.
# This way we are sure it will not finish before we test
# the state while it is being rebuilt.
with mock.patch.object(GstTranscoder.Transcoder, "run_async"):
mainloop, app, medialib, proxy_uri = self.load_project_with_missing_proxy()
uri = common.get_sample_uri("1sec_simpsons_trailer.mp4")
app.project_manager.connect("new-project-loaded", lambda x, y: mainloop.quit())
mainloop.run()
app.project_manager.connect("new-project-loaded", lambda x, y: mainloop.quit())
mainloop.run()
self.assertEqual(len(medialib.storemodel), 1)
self.assertEqual(medialib.storemodel[0][medialibrary.COL_ASSET].props.id,
uri)
self.assertEqual(medialib.storemodel[0][medialibrary.COL_THUMB_DECORATOR].state,
medialibrary.AssetThumbnail.IN_PROGRESS)
self.assertEqual(len(medialib.storemodel), 1)
self.assertEqual(medialib.storemodel[0][medialibrary.COL_ASSET].props.id,
uri)
self.assertEqual(medialib.storemodel[0][medialibrary.COL_THUMB_DECORATOR].state,
medialibrary.AssetThumbnail.IN_PROGRESS)
def test_load_project_with_missing_proxy_stop_generating_and_proxy(self):
"""Checks cancelling creation of a missing proxies and forcing it again."""
from gi.repository import GstTranscoder
# Disable proxy generation by not making it start ever.
# This way we are sure it will not finish before we test
# stop generating the proxy and restart it.
with mock.patch.object(GstTranscoder.Transcoder, "run_async"):
mainloop, app, medialib, proxy_uri = self.load_project_with_missing_proxy()
uri = common.get_sample_uri("1sec_simpsons_trailer.mp4")
with common.cloned_sample("1sec_simpsons_trailer.mp4"):
# Disable proxy generation by not making it start ever.
# This way we are sure it will not finish before we test
# stop generating the proxy and restart it.
with mock.patch.object(GstTranscoder.Transcoder, "run_async"):
mainloop, app, medialib, proxy_uri = self.load_project_with_missing_proxy()
app.project_manager.connect("new-project-loaded", lambda x, y: mainloop.quit())
mainloop.run()
asset = medialib.storemodel[0][medialibrary.COL_ASSET]
app.project_manager.current_project.disable_proxies_for_assets([asset])
app.project_manager.connect("new-project-loaded", lambda x, y: mainloop.quit())
mainloop.run()
asset = medialib.storemodel[0][medialibrary.COL_ASSET]
app.project_manager.current_project.disable_proxies_for_assets([asset])
row, = medialib.storemodel
asset = row[medialibrary.COL_ASSET]
self.assertEqual(medialib._progressbar.get_fraction(), 1.0)
uri = common.get_sample_uri("1sec_simpsons_trailer.mp4")
self.assertEqual(asset.props.id, uri)
self.assertEqual(asset.ready, True)
self.assertEqual(asset.creation_progress, 100)
self.assertEqual(row[medialibrary.COL_THUMB_DECORATOR].state,
medialibrary.AssetThumbnail.NO_PROXY)
app.project_manager.current_project.use_proxies_for_assets([asset])
mainloop.run()
app.project_manager.current_project.use_proxies_for_assets([asset])
mainloop.run()
row, = medialib.storemodel
asset = row[medialibrary.COL_ASSET]
......@@ -494,9 +498,6 @@ class TestProjectLoading(common.TestCase):
mainloop = common.create_main_loop()
# Remove proxy
common.clean_proxy_samples()
def new_project_loaded_cb(*args, **kwargs):
mainloop.quit()
......@@ -507,7 +508,8 @@ class TestProjectLoading(common.TestCase):
project_manager.connect("new-project-loaded", new_project_loaded_cb)
project_manager.loadProject(proj_uri)
mainloop.run()
with common.cloned_sample("1sec_simpsons_trailer.mp4"):
mainloop.run()
self.assertEqual(medialib._progressbar.get_fraction(), 1.0)
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment