123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184 |
- #!/usr/bin/env vpython3
- # Copyright 2020 The Chromium Authors. All rights reserved.
- # Use of this source code is governed by a BSD-style license that can be
- # found in the LICENSE file.
- #pylint: disable=protected-access
- import json
- import os
- import sys
- import tempfile
- import typing
- import unittest
- if sys.version_info[0] == 2:
- import mock
- else:
- import unittest.mock as mock
- from pyfakefs import fake_filesystem_unittest
- from skia_gold_common import skia_gold_properties
- from skia_gold_common import skia_gold_session
- from skia_gold_common import skia_gold_session_manager
- from skia_gold_common import unittest_utils
- createSkiaGoldArgs = unittest_utils.createSkiaGoldArgs
- class SkiaGoldSessionManagerGetSessionTest(fake_filesystem_unittest.TestCase):
- """Tests the functionality of SkiaGoldSessionManager.GetSkiaGoldSession."""
- def setUp(self) -> None:
- self.setUpPyfakefs()
- self._working_dir = tempfile.mkdtemp()
- self._patcher = mock.patch.object(
- skia_gold_session_manager.SkiaGoldSessionManager, 'GetSessionClass')
- self._session_class_mock = self._patcher.start()
- self._session_class_mock.return_value = skia_gold_session.SkiaGoldSession
- self.addCleanup(self._patcher.stop)
- def test_ArgsForwardedToSession(self) -> None:
- args = createSkiaGoldArgs()
- sgp = skia_gold_properties.SkiaGoldProperties(args)
- sgsm = skia_gold_session_manager.SkiaGoldSessionManager(
- self._working_dir, sgp)
- session = sgsm.GetSkiaGoldSession({}, 'corpus', 'instance')
- self.assertTrue(session._keys_file.startswith(self._working_dir))
- self.assertEqual(session._corpus, 'corpus')
- self.assertEqual(session._instance, 'instance')
- # Make sure the session's working directory is a subdirectory of the
- # manager's working directory.
- self.assertEqual(os.path.dirname(session._working_dir), self._working_dir)
- def test_corpusFromJson(self) -> None:
- args = createSkiaGoldArgs()
- sgp = skia_gold_properties.SkiaGoldProperties(args)
- sgsm = skia_gold_session_manager.SkiaGoldSessionManager(
- self._working_dir, sgp)
- session = sgsm.GetSkiaGoldSession({'source_type': 'foobar'}, None,
- 'instance')
- self.assertTrue(session._keys_file.startswith(self._working_dir))
- self.assertEqual(session._corpus, 'foobar')
- self.assertEqual(session._instance, 'instance')
- def test_corpusDefaultsToInstance(self) -> None:
- args = createSkiaGoldArgs()
- sgp = skia_gold_properties.SkiaGoldProperties(args)
- sgsm = skia_gold_session_manager.SkiaGoldSessionManager(
- self._working_dir, sgp)
- session = sgsm.GetSkiaGoldSession({}, None, 'instance')
- self.assertTrue(session._keys_file.startswith(self._working_dir))
- self.assertEqual(session._corpus, 'instance')
- self.assertEqual(session._instance, 'instance')
- @mock.patch.object(skia_gold_session_manager.SkiaGoldSessionManager,
- '_GetDefaultInstance')
- def test_getDefaultInstance(self,
- default_instance_mock: mock.MagicMock) -> None:
- default_instance_mock.return_value = 'default'
- args = createSkiaGoldArgs()
- sgp = skia_gold_properties.SkiaGoldProperties(args)
- sgsm = skia_gold_session_manager.SkiaGoldSessionManager(
- self._working_dir, sgp)
- session = sgsm.GetSkiaGoldSession({}, None, None)
- self.assertTrue(session._keys_file.startswith(self._working_dir))
- self.assertEqual(session._corpus, 'default')
- self.assertEqual(session._instance, 'default')
- @mock.patch.object(skia_gold_session.SkiaGoldSession, '__init__')
- def test_matchingSessionReused(self, session_mock: mock.MagicMock) -> None:
- session_mock.return_value = None
- args = createSkiaGoldArgs()
- sgp = skia_gold_properties.SkiaGoldProperties(args)
- sgsm = skia_gold_session_manager.SkiaGoldSessionManager(
- self._working_dir, sgp)
- session1 = sgsm.GetSkiaGoldSession({}, 'corpus', 'instance')
- session2 = sgsm.GetSkiaGoldSession({}, 'corpus', 'instance')
- self.assertEqual(session1, session2)
- # For some reason, session_mock.assert_called_once() always passes,
- # so check the call count directly.
- self.assertEqual(session_mock.call_count, 1)
- @mock.patch.object(skia_gold_session.SkiaGoldSession, '__init__')
- def test_separateSessionsFromKeys(self, session_mock: mock.MagicMock) -> None:
- session_mock.return_value = None
- args = createSkiaGoldArgs()
- sgp = skia_gold_properties.SkiaGoldProperties(args)
- sgsm = skia_gold_session_manager.SkiaGoldSessionManager(
- self._working_dir, sgp)
- session1 = sgsm.GetSkiaGoldSession({}, 'corpus', 'instance')
- session2 = sgsm.GetSkiaGoldSession({'something_different': 1}, 'corpus',
- 'instance')
- self.assertNotEqual(session1, session2)
- self.assertEqual(session_mock.call_count, 2)
- @mock.patch.object(skia_gold_session.SkiaGoldSession, '__init__')
- def test_separateSessionsFromCorpus(self,
- session_mock: mock.MagicMock) -> None:
- session_mock.return_value = None
- args = createSkiaGoldArgs()
- sgp = skia_gold_properties.SkiaGoldProperties(args)
- sgsm = skia_gold_session_manager.SkiaGoldSessionManager(
- self._working_dir, sgp)
- session1 = sgsm.GetSkiaGoldSession({}, 'corpus1', 'instance')
- session2 = sgsm.GetSkiaGoldSession({}, 'corpus2', 'instance')
- self.assertNotEqual(session1, session2)
- self.assertEqual(session_mock.call_count, 2)
- @mock.patch.object(skia_gold_session.SkiaGoldSession, '__init__')
- def test_separateSessionsFromInstance(self,
- session_mock: mock.MagicMock) -> None:
- session_mock.return_value = None
- args = createSkiaGoldArgs()
- sgp = skia_gold_properties.SkiaGoldProperties(args)
- self._working_dir = tempfile.mkdtemp()
- sgsm = skia_gold_session_manager.SkiaGoldSessionManager(
- self._working_dir, sgp)
- session1 = sgsm.GetSkiaGoldSession({}, 'corpus', 'instance1')
- session2 = sgsm.GetSkiaGoldSession({}, 'corpus', 'instance2')
- self.assertNotEqual(session1, session2)
- self.assertEqual(session_mock.call_count, 2)
- class SkiaGoldSessionManagerKeyConversionTest(fake_filesystem_unittest.TestCase
- ):
- def setUp(self) -> None:
- self.setUpPyfakefs()
- self._working_dir = tempfile.mkdtemp()
- def test_getKeysAsDict(self) -> None:
- keys_dict = {'foo': 'bar'}
- keys_file_contents = {'bar': 'baz'}
- keys_file = tempfile.NamedTemporaryFile(delete=False).name
- with open(keys_file, 'w') as f:
- json.dump(keys_file_contents, f)
- self.assertEqual(skia_gold_session_manager._GetKeysAsDict(keys_dict),
- keys_dict)
- self.assertEqual(skia_gold_session_manager._GetKeysAsDict(keys_file),
- keys_file_contents)
- with self.assertRaises(AssertionError):
- skia_gold_session_manager._GetKeysAsDict(typing.cast(dict, 1))
- def test_getKeysAsJson(self) -> None:
- keys_dict = {'foo': 'bar'}
- keys_file_contents = {'bar': 'baz'}
- keys_file = tempfile.NamedTemporaryFile(delete=False).name
- with open(keys_file, 'w') as f:
- json.dump(keys_file_contents, f)
- self.assertEqual(skia_gold_session_manager._GetKeysAsJson(keys_file, ''),
- keys_file)
- keys_dict_as_json = skia_gold_session_manager._GetKeysAsJson(
- keys_dict, self._working_dir)
- self.assertTrue(keys_dict_as_json.startswith(self._working_dir))
- with open(keys_dict_as_json) as f:
- self.assertEqual(json.load(f), keys_dict)
- with self.assertRaises(AssertionError):
- skia_gold_session_manager._GetKeysAsJson(typing.cast(dict, 1), '')
- if __name__ == '__main__':
- unittest.main(verbosity=2)
|