KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use test_sync method in autotest

Best Python code snippet using autotest_python

test_watcher.py

Source:test_watcher.py

1# (c) 2014 Amplify Education, Inc. All rights reserved, subject to the license2# below.3#4# Education agencies that are members of the Smarter Balanced Assessment5# Consortium as of August 1, 2014 are granted a worldwide, non-exclusive, fully6# paid-up, royalty-free, perpetual license, to access, use, execute, reproduce,7# display, distribute, perform and create derivative works of the software8# included in the Reporting Platform, including the source code to such software.9# This license includes the right to grant sublicenses by such consortium members10# to third party vendors solely for the purpose of performing services on behalf11# of such consortium member educational agencies.12import unittest13import shutil14import tempfile15import time16import os17from edcore.tests.watch.common_test_utils import write_something_to_a_blank_file, create_checksum_file18from edcore.watch.file_hasher import MD5Hasher19from edcore.watch.watcher import FileWatcher20class TestWatcher(unittest.TestCase):21    @classmethod22    def setUpClass(cls):23        pass24    @classmethod25    def tearDownClass(cls):26        pass27    def setUp(self):28        self.pattern = "['*.gpg', '*.gpg.done']"29        self.base_dir = tempfile.mkdtemp(prefix='base')30        self.source_dir = tempfile.mkdtemp(prefix='arrivals', dir=self.base_dir)31        self.source_path = os.path.join(self.base_dir, self.source_dir)32        self.conf = {33            'base_dir': self.base_dir,34            'source_dir': self.source_dir,35            'file_patterns_to_watch': self.pattern,36            'file_stat_watch_interval': 1,37            'file_stat_watch_period': 6,38            'file_checksum_threshold_wait_period': 4,39            'file_system_scan_delay': 240        }41        self.test_sync = FileWatcher(self.conf)42        self.test_hasher = MD5Hasher()43        self.tmp_dir_1 = tempfile.mkdtemp(prefix='tmp_1', dir=self.source_path)44        self.tmp_dir_2 = tempfile.mkdtemp(prefix='tmp_2', dir=self.source_path)45        self.test_file_1 = tempfile.NamedTemporaryFile(delete=False, suffix='*.gpg',46                                                       prefix='test_file_1', dir=self.tmp_dir_1)47        self.test_file_2 = tempfile.NamedTemporaryFile(delete=False, suffix='*.gpg',48                                                       prefix='test_file_2', dir=self.tmp_dir_2)49    def _without_timestamps(self, file_stats):50        dict_without_timestamps = {}51        for file in file_stats.keys():52            dict_without_timestamps[file] = file_stats[file][0]53        return dict_without_timestamps54    def tearDown(self):55        shutil.rmtree(self.base_dir, ignore_errors=True)56    def test_clear_file_stats(self):57        self.test_sync.clear_file_stats()58        self.assertEqual(self._without_timestamps(self.test_sync.get_file_stats()), {})59    def test_find_all_files(self):60        self.test_sync.find_all_files()61        self.assertEqual(self._without_timestamps(self.test_sync.get_file_stats()), {62            self.test_file_1.name: 0, self.test_file_2.name: 0})63    def test_add_file_to_snapshot(self):64        self.assertEqual(len(self.test_sync.get_file_stats().keys()), 0)65        self.test_sync.add_file_to_snapshot(self.test_file_1.name)66        self.assertEqual(len(self.test_sync.get_file_stats().keys()), 1)67        self.assertEqual(self._without_timestamps(self.test_sync.get_file_stats()), {self.test_file_1.name: 0})68    def test_get_updated_file_stats(self):69        self.test_sync.find_all_files()70        self.assertEqual(self._without_timestamps(self.test_sync.get_file_stats()), {71            self.test_file_1.name: 0, self.test_file_2.name: 0})72        self.test_file_1.write(b"test\n")73        self.test_file_1.flush()74        self.assertEqual(self._without_timestamps(self.test_sync.get_updated_file_stats()), {75            self.test_file_1.name: 5, self.test_file_2.name: 0})76    def test_watch_files_1(self):77        self.test_sync.find_all_files()78        self.assertEqual(self._without_timestamps(self.test_sync.get_file_stats()), {79            self.test_file_1.name: 0, self.test_file_2.name: 0})80        stop = self.test_sync.watch_and_filter_files_by_stats_changes()81        time.sleep(2)82        self.test_file_1.write(b"test\n")83        self.test_file_1.flush()84        time.sleep(self.conf['file_stat_watch_period'])85        stop.set()86        self.assertEqual(self._without_timestamps(self.test_sync.get_file_stats()), {self.test_file_2.name: 0})87    def test_watch_files_2(self):88        self.test_sync.find_all_files()89        self.assertEqual(self._without_timestamps(self.test_sync.get_file_stats()), {90            self.test_file_1.name: 0, self.test_file_2.name: 0})91        stop = self.test_sync.watch_and_filter_files_by_stats_changes()92        time.sleep(2)93        self.test_file_1.write(b"test\n")94        self.test_file_1.flush()95        time.sleep(self.conf['file_stat_watch_period'])96        stop.set()97        self.assertEqual(self._without_timestamps(self.test_sync.get_file_stats()), {self.test_file_2.name: 0})98    def test_valid_check_sum_with_no_checksum_file(self):99        test_file_path = write_something_to_a_blank_file(dir_path=self.tmp_dir_1)100        self.assertFalse(self.test_sync.valid_check_sum(test_file_path))101    def test_valid_check_sum_with_no_checksum_file_for_shorter_time(self):102        test_file_path = write_something_to_a_blank_file(dir_path=self.tmp_dir_1)103        time.sleep(2)104        self.assertFalse(self.test_sync.valid_check_sum(test_file_path))105    def test_valid_check_sum_with_no_checksum_file_for_longer_time(self):106        test_file_path = write_something_to_a_blank_file(dir_path=self.tmp_dir_1)107        time.sleep(5)108        self.assertFalse(self.test_sync.valid_check_sum(test_file_path))109    def test_valid_check_sum_with_valid_checksum_file(self):110        test_file_path = write_something_to_a_blank_file(dir_path=self.tmp_dir_1)111        _ = create_checksum_file(test_file_path)112        self.assertTrue(self.test_sync.valid_check_sum(test_file_path))113    def test_valid_check_sum_with_invalid_checksum_file(self):114        test_file_path = write_something_to_a_blank_file(dir_path=self.tmp_dir_1)115        _ = create_checksum_file(test_file_path, valid_check_sum=False)116        self.assertFalse(self.test_sync.valid_check_sum(test_file_path))117    def test_remove_files_from_dict(self):118        test_file_path = write_something_to_a_blank_file(dir_path=self.tmp_dir_1)119        checksum_file_path = create_checksum_file(test_file_path)120        self.test_sync.find_all_files()121        self.assertEqual(self._without_timestamps(self.test_sync.get_file_stats()), {122            self.test_file_1.name: 0, self.test_file_2.name: 0,123            test_file_path: 5, checksum_file_path: 37})124        self.test_sync.remove_file_from_dict(self.test_file_1.name)125        self.assertEqual(self._without_timestamps(self.test_sync.get_file_stats()), {126            self.test_file_2.name: 0, test_file_path: 5, checksum_file_path: 37})127        self.test_sync.remove_file_pair_from_dict(test_file_path)128        self.assertEqual(self._without_timestamps(self.test_sync.get_file_stats()), {self.test_file_2.name: 0})129        self.test_sync.remove_file_pair_from_dict(self.test_file_2.name)130        self.assertEqual(self._without_timestamps(self.test_sync.get_file_stats()), {})131    def test_filter_files_for_digest_mismatch_1(self):132        test_file_3_path = write_something_to_a_blank_file(dir_path=self.tmp_dir_1)133        checksum_file_3_path = create_checksum_file(test_file_3_path)134        test_file_4_path = write_something_to_a_blank_file(dir_path=self.tmp_dir_1)135        checksum_file_4_path = create_checksum_file(test_file_4_path)136        self.test_sync.find_all_files()137        self.assertEqual(self._without_timestamps(self.test_sync.get_file_stats()),138                         {self.test_file_1.name: 0, self.test_file_2.name: 0,139                          test_file_3_path: 5, checksum_file_3_path: 37,140                          test_file_4_path: 5, checksum_file_4_path: 37})141        self.test_sync.filter_files_for_digest_mismatch()142        self.assertEqual(self._without_timestamps(self.test_sync.get_file_stats()),143                         {test_file_3_path: 5, checksum_file_3_path: 37,144                          test_file_4_path: 5, checksum_file_4_path: 37})145        test_file_5_path = write_something_to_a_blank_file(dir_path=self.tmp_dir_1)146        checksum_file_5_path = create_checksum_file(test_file_5_path, valid_check_sum=False)147        self.test_sync.find_all_files()148        self.assertEqual(self._without_timestamps(self.test_sync.get_file_stats()), {149            self.test_file_1.name: 0, self.test_file_2.name: 0,150            test_file_3_path: 5, checksum_file_3_path: 37,151            test_file_4_path: 5, checksum_file_4_path: 37,152            test_file_5_path: 5, checksum_file_5_path: 26})153        self.test_sync.filter_files_for_digest_mismatch()154        self.assertEqual(self._without_timestamps(self.test_sync.get_file_stats()), {155            test_file_3_path: 5, checksum_file_3_path: 37,156            test_file_4_path: 5, checksum_file_4_path: 37})157    def test_filter_files_for_digest_mismatch_2(self):158        test_file_3_path = write_something_to_a_blank_file(dir_path=self.tmp_dir_1)159        checksum_file_3_path = create_checksum_file(test_file_3_path)160        test_file_4_path = write_something_to_a_blank_file(dir_path=self.tmp_dir_1)161        checksum_file_4_path = create_checksum_file(test_file_4_path)162        self.test_sync.find_all_files()163        self.assertEqual(self._without_timestamps(self.test_sync.get_file_stats()), {164            self.test_file_1.name: 0, self.test_file_2.name: 0,165            test_file_3_path: 5, checksum_file_3_path: 37,166            test_file_4_path: 5, checksum_file_4_path: 37})167        self.test_sync.filter_checksum_files()168        self.assertEqual(self._without_timestamps(self.test_sync.get_file_stats()), {169            self.test_file_1.name: 0, self.test_file_2.name: 0,170            test_file_3_path: 5, test_file_4_path: 5})171    def test_handle_missing_checksum_files_when_user_forgets_to_drop_checksum_file(self):172        self.test_sync.find_all_files()173        self.assertEqual(self._without_timestamps(self.test_sync.get_file_stats()),174                         {self.test_file_1.name: 0, self.test_file_2.name: 0})175        self.test_sync.handle_missing_checksum_files()176        self.assertEqual(len(self._without_timestamps(self.test_sync.get_file_stats()).keys()), 4)177        self.assertEqual(set(self._without_timestamps(self.test_sync.get_file_stats()).keys()),178                         {self.test_file_1.name, self.test_file_2.name,179                          self.test_file_1.name + '.done', self.test_file_2.name + '.done'})180    def test_handle_missing_checksum_files_with_both_missing_and_avail(self):181        test_file_3_path = write_something_to_a_blank_file(dir_path=self.tmp_dir_1)182        checksum_file_3_path = create_checksum_file(test_file_3_path)183        self.test_sync.find_all_files()184        self.assertEqual(self._without_timestamps(self.test_sync.get_file_stats()),185                         {self.test_file_1.name: 0, self.test_file_2.name: 0,186                          test_file_3_path: 5, checksum_file_3_path: 37})187        self.test_sync.handle_missing_checksum_files()188        self.assertEqual(len(self._without_timestamps(self.test_sync.get_file_stats()).keys()), 6)189        self.assertEqual(set(self._without_timestamps(self.test_sync.get_file_stats()).keys()),190                         {self.test_file_1.name, self.test_file_2.name, test_file_3_path,191                          self.test_file_1.name + '.done', self.test_file_2.name + '.done', test_file_3_path + '.done'})192    def test_handle_missing_checksum_files_when_user_drops_checksum_file_during_a_watch_period(self):193        test_file_3_path = write_something_to_a_blank_file(dir_path=self.tmp_dir_1)194        self.test_sync.find_all_files()195        self.assertEqual(self._without_timestamps(self.test_sync.get_file_stats()),196                         {self.test_file_1.name: 0, self.test_file_2.name: 0,197                          test_file_3_path: 5})198        # user drops the checksum file after a snapshot is taken199        checksum_file_3_path = create_checksum_file(test_file_3_path)200        self.test_sync.watch_files()201        self.test_sync.handle_missing_checksum_files()202        self.assertEqual(len(self._without_timestamps(self.test_sync.get_file_stats()).keys()), 4)203        self.assertEqual(set(self._without_timestamps(self.test_sync.get_file_stats()).keys()),204                         {self.test_file_1.name, self.test_file_2.name,...

test_kosh_sync.py

Source:test_kosh_sync.py

1import os2from koshbase import KoshTest3import time4from sina.utils import DataRange5class KoshTestSync(KoshTest):6    def test_sync_mode_switch(self):7        store, kosh_db = self.connect()8        self.assertTrue(store.is_synchronous())9        self.assertTrue(store.__sync__)10        self.assertFalse(store.synchronous())11        self.assertFalse(store.__sync__)12        self.assertFalse(store.is_synchronous())13        self.assertTrue(store.synchronous())14        self.assertTrue(store.__sync__)15        self.assertTrue(store.is_synchronous())16        self.assertTrue(store.synchronous(True))17        self.assertTrue(store.__sync__)18        self.assertTrue(store.is_synchronous())19        self.assertFalse(store.synchronous(False))20        self.assertFalse(store.__sync__)21        self.assertFalse(store.is_synchronous())22        self.assertFalse(store.synchronous(False))23        self.assertFalse(store.__sync__)24        self.assertTrue(store.synchronous(True))25        self.assertTrue(store.__sync__)26        self.assertTrue(store.is_synchronous())27        os.remove(kosh_db)28    def test_sync_associate_fail(self):29        store, kosh_db = self.connect()30        store2, kosh_db = self.connect(db_uri=kosh_db, sync=False)31        # Create many datasets32        ds = store.create(metadata={"key1": 1, "key2": "A"})33        ds2 = store2.create(metadata={"key2": "B", "key3": 3})34        ds3 = store.create()35        ds.associate("tests/baselines/node_extracts2", "something")36        ds2.associate("tests/baselines/node_extracts2", "something")37        ds3.associate("tests/baselines/node_extracts2", "something")38        # associated record was modified since ds2 modified it39        # Cannot sync any longer40        with self.assertRaises(RuntimeError):41            store2.sync()42        store.close()43        store2.close()44        os.remove(kosh_db)45    def test_sync_find(self):46        store, kosh_db = self.connect()47        store2, kosh_db = self.connect(db_uri=kosh_db, sync=False)48        # Create many datasets49        ds = store.create(metadata={"key1": 1, "key2": "A"})50        ds2 = store2.create(metadata={"key2": "B", "key3": 3})51        ds3 = store.create()52        store2.create(metadata={"key2": "C", "key3": 4})53        ds.associate("tests/baselines/node_extracts2", "something")54        ds3.associate("tests/baselines/node_extracts2", "something")55        ds2.associate("tests/baselines/node_extracts2", "something")56        s = list(store.find(key2=DataRange("A")))57        self.assertEqual(len(s), 1)58        s = list(store2.find(key2=DataRange("A")))59        self.assertEqual(len(s), 3)60        s = list(store.find(61            key2=DataRange("A"),62            file=os.path.abspath("tests/baselines/node_extracts2")))63        self.assertEqual(len(s), 1)64        s = list(store2.find(65            key2=DataRange("A"),66            file=os.path.abspath("tests/baselines/node_extracts2")))67        self.assertEqual(len(s), 2)68        store2.sync()69        s = list(store.find(key2=DataRange("A")))70        self.assertEqual(len(s), 3)71        s = list(store2.find(key2=DataRange("A")))72        self.assertEqual(len(s), 3)73        s = list(store.find(74            key2=DataRange("A"),75            file=os.path.abspath("tests/baselines/node_extracts2")))76        self.assertEqual(len(s), 2)77        s = list(store2.find(78            key2=DataRange("A"),79            file=os.path.abspath("tests/baselines/node_extracts2")))80        self.assertEqual(len(s), 2)81        store.close()82        store2.close()83        os.remove(kosh_db)84    def test_sync_delete_dataset(self):85        store1, kosh_db = self.connect(sync=True)86        store2, kosh_db = self.connect(db_uri=kosh_db, sync=False)87        # Create dataset on syncing store88        ds1 = store1.create()89        dsid = ds1.id90        # Check it exists on store291        self.assertEqual(len(list(store1.find())), 1)92        self.assertEqual(len(list(store2.find())), 1)93        store2.delete(dsid)94        # self.assertEqual(len(store2.find()),0)95        store2.sync()96        self.assertEqual(len(list(store1.find())), 0)97        store2, kosh_db = self.connect(db_uri=kosh_db)98        with self.assertRaises(Exception):99            store2.open(dsid)100        self.assertEqual(len(list(store2.find())), 0)101        store1.close()102        store2.close()103        os.remove(kosh_db)104    def test_sync_dataset_attributes(self):105        store1, kosh_db = self.connect(sync=True)106        store2, kosh_db = self.connect(db_uri=kosh_db, sync=False)107        # Create dataset on syncing store108        ds1 = store1.create()109        ds1.test_sync = "Set"110        # Check it exists on store2111        ds2 = store2.open(ds1.id)112        # Check they are identical113        self.assertEqual(ds2.id, ds1.id)114        self.assertEqual(ds2.test_sync, ds1.test_sync)115        # Change in store1, shouldn't change on store2 until synced116        ds1.test_sync = "Changed"117        self.assertEqual(ds1.test_sync, "Changed")118        self.assertNotEqual(ds1.test_sync, ds2.test_sync)119        self.assertEqual(ds2.test_sync, "Set")120        # Sync dataset121        ds2.sync()122        self.assertEqual(ds2.test_sync, ds1.test_sync)123        self.assertEqual(ds1.test_sync, "Changed")124        # Another change125        ds2.test_sync = "Changed from 2nd store"126        self.assertEqual(ds2.test_sync, "Changed from 2nd store")127        self.assertNotEqual(ds1.test_sync, ds2.test_sync)128        self.assertEqual(ds1.test_sync, "Changed")129        ds2.sync()130        self.assertEqual(ds2.test_sync, ds1.test_sync)131        self.assertEqual(ds2.test_sync, "Changed from 2nd store")132        self.assertEqual(ds1.test_sync, "Changed from 2nd store")133        # Now change on store134        ds3 = store2.create()135        ds3.test_sync = "exists"136        # Check it does not exists on store1137        with self.assertRaises(Exception):138            ds3 = store1.open(ds3.id)139        ds2.test_sync = "Another change"140        self.assertNotEqual(ds1.test_sync, ds2.test_sync)141        # Sync the store142        store2.sync()143        self.assertEqual(ds2.test_sync, ds1.test_sync)144        self.assertEqual(ds1.test_sync, "Another change")145        ds3 = store1.open(ds3.id)146        self.assertEqual(ds3.test_sync, "exists")147        # ok now test it fails if store changed in between148        ds2.test_sync = "I changed it"149        time.sleep(.1)150        ds1.test_sync = "I changed it after you"151        with self.assertRaises(RuntimeError):152            ds2.sync()153        with self.assertRaises(RuntimeError):154            ds2.sync()155        self.assertEqual(ds2.test_sync, "I changed it")156        with self.assertRaises(RuntimeError):157            ds2.sync()158        self.assertEqual(ds1.test_sync, "I changed it after you")159        with self.assertRaises(RuntimeError):160            ds2.sync()161        ds2.test_sync = ds1.test_sync162        ds2.sync()163        self.assertEqual(ds1.test_sync, "I changed it after you")164        self.assertEqual(ds2.test_sync, "I changed it after you")165        # Now testing deletion stuff166        del(ds1.test_sync)167        ds2.test_sync = "Ok let's change you"168        with self.assertRaises(RuntimeError):169            ds2.sync()170        del(ds2.test_sync)171        ds2.sync()172        ds2.associate("ghost", "not_real")173        self.assertNotEqual(ds2._associated_data_, ds1._associated_data_)174        ds1.associate("ghostly", "fake")175        ds2.associate("ghostlier", "not_real_as_well")176        self.assertEqual(len(ds2._associated_data_), 2)177        ds2.sync()178        self.assertEqual(len(ds2._associated_data_), 3)179        self.assertEqual(ds2._associated_data_, ds1._associated_data_)180        ds2.dissociate("ghost")181        self.assertEqual(len(ds2._associated_data_), 2)182        self.assertEqual(len(ds1._associated_data_), 3)183        ds2.sync()184        self.assertEqual(len(ds1._associated_data_), 2)185        self.assertEqual(len(ds2._associated_data_), 2)186        # Ok now let's see if we do conflict187        ds2.associate("conflict", "conf")188        ds1.associate("conflict", "conf2")189        with self.assertRaises(RuntimeError):190            ds2.sync()191        ds2.dissociate("conflict")192        ds2.associate("conflict", "conf2")193        ds2.sync()194        store2.sync()195        store1.close()196        store2.close()197        os.remove(kosh_db)198if __name__ == "__main__":199    A = KoshTestSync()200    for nm in dir(A):201        if nm[:4] == "test":202            fn = getattr(A, nm)203            print(nm, fn)...

sync_widget_test.py

Source:sync_widget_test.py

1import unittest2from PySide6.QtCore import QSize3from src.model.main_model import MainModel4from src.view.widgets.sync_widget import SyncWidget5from src.controllers.widgets.sync_controller import SyncController6from tests import default_code7class SyncWidgetTest(default_code.DefaultCode):8    """ Test synchronized widget class"""9    def setUp(self):10        """Metodo che viene chiamato prima di ogni metodo"""11        super().setUp()12        self.main_model = MainModel()13        self.sync_model = self.main_model.sync_model14        self.test_sync = SyncWidget(self.sync_model)15        self.controller = SyncController(self.sync_model, self.test_sync)16    def tearDown(self):17        """Metodo che viene chiamato dopo ogni metodo"""18        super().tearDown()19    def test_defaults(self):20        """Test default synchronized widget"""21        self.assertEqual(self.test_sync.watch_label.text(), "SYNC")22        self.assertEqual(self.test_sync.watch_label.accessibleName(), "Title")23        self.assertEqual(self.test_sync.running_label.text(), "Disattivata")24        self.assertEqual(self.test_sync.running_label.accessibleName(), "Subtitle")25        self.assertEqual(self.test_sync.sync_button.iconSize(), QSize(50, 50))26        self.assertTrue(self.test_sync.sync_button.isCheckable())27        self.assertEqual(self.test_sync.sync_button.accessibleName(), "HighlightButton")28        self.assertEqual(self.test_sync.menu_label.text(), "â¢ â¢ â¢")29    def test_turn_on_sync(self):30        self.test_sync.sync_button.click()31        self.assertEqual(self.test_sync.sync_button.isChecked(), self.test_sync._model.get_state())32        self.assertEqual(self.test_sync.running_label.text(), "Attivata")33    def test_turn_off_sync(self):34        self.test_sync.sync_button.click()35        self.assertEqual(self.test_sync.sync_button.isChecked(), self.test_sync._model.get_state())36        self.assertEqual(self.test_sync.running_label.text(), "Attivata")37        self.test_sync.sync_button.click()38        self.assertEqual(self.test_sync.sync_button.isChecked(), self.test_sync._model.get_state())39        self.assertEqual(self.test_sync.running_label.text(), "Disattivata")40if __name__ == "__main__":...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.