Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use ok method in Sure

Best Python code snippet using sure_python

ldb_test.py

Source:ldb_test.py

1import os2import os.path3import shutil4import subprocess5import time6import unittest7import tempfile8def my_check_output(*popenargs, **kwargs):9    """10    If we had python 2.7, we should simply use subprocess.check_output.11    This is a stop-gap solution for python 2.612    """13    if 'stdout' in kwargs:14        raise ValueError('stdout argument not allowed, it will be overridden.')15    process = subprocess.Popen(stderr=subprocess.PIPE, stdout=subprocess.PIPE,16                               *popenargs, **kwargs)17    output, unused_err = process.communicate()18    retcode = process.poll()19    if retcode:20        cmd = kwargs.get("args")21        if cmd is None:22            cmd = popenargs[0]23        raise Exception("Exit code is not 0.  It is %d.  Command: %s" %24                (retcode, cmd))25    return output26def run_err_null(cmd):27    return os.system(cmd + " 2>/dev/null ")28class LDBTestCase(unittest.TestCase):29    def setUp(self):30        self.TMP_DIR  = tempfile.mkdtemp(prefix="ldb_test_")31        self.DB_NAME = "testdb"32    def tearDown(self):33        assert(self.TMP_DIR.strip() != "/"34                and self.TMP_DIR.strip() != "/tmp"35                and self.TMP_DIR.strip() != "/tmp/") #Just some paranoia36        shutil.rmtree(self.TMP_DIR)37    def dbParam(self, dbName):38        return "--db=%s" % os.path.join(self.TMP_DIR, dbName)39    def assertRunOKFull(self, params, expectedOutput, unexpected=False):40        """41        All command-line params must be specified.42        Allows full flexibility in testing; for example: missing db param.43        """44        output = my_check_output("./ldb %s |grep -v \"Created bg thread\"" %45                            params, shell=True)46        if not unexpected:47            self.assertEqual(output.strip(), expectedOutput.strip())48        else:49            self.assertNotEqual(output.strip(), expectedOutput.strip())50    def assertRunFAILFull(self, params):51        """52        All command-line params must be specified.53        Allows full flexibility in testing; for example: missing db param.54        """55        try:56            my_check_output("./ldb %s >/dev/null 2>&1 |grep -v \"Created bg \57                thread\"" % params, shell=True)58        except Exception, e:59            return60        self.fail(61            "Exception should have been raised for command with params: %s" %62            params)63    def assertRunOK(self, params, expectedOutput, unexpected=False):64        """65        Uses the default test db.66        """67        self.assertRunOKFull("%s %s" % (self.dbParam(self.DB_NAME), params),68                             expectedOutput, unexpected)69    def assertRunFAIL(self, params):70        """71        Uses the default test db.72        """73        self.assertRunFAILFull("%s %s" % (self.dbParam(self.DB_NAME), params))74    def testSimpleStringPutGet(self):75        print "Running testSimpleStringPutGet..."76        self.assertRunFAIL("put x1 y1")77        self.assertRunOK("put --create_if_missing x1 y1", "OK")78        self.assertRunOK("get x1", "y1")79        self.assertRunFAIL("get x2")80        self.assertRunOK("put x2 y2", "OK")81        self.assertRunOK("get x1", "y1")82        self.assertRunOK("get x2", "y2")83        self.assertRunFAIL("get x3")84        self.assertRunOK("scan --from=x1 --to=z", "x1 : y1\nx2 : y2")85        self.assertRunOK("put x3 y3", "OK")86        self.assertRunOK("scan --from=x1 --to=z", "x1 : y1\nx2 : y2\nx3 : y3")87        self.assertRunOK("scan", "x1 : y1\nx2 : y2\nx3 : y3")88        self.assertRunOK("scan --from=x", "x1 : y1\nx2 : y2\nx3 : y3")89        self.assertRunOK("scan --to=x2", "x1 : y1")90        self.assertRunOK("scan --from=x1 --to=z --max_keys=1", "x1 : y1")91        self.assertRunOK("scan --from=x1 --to=z --max_keys=2",92                "x1 : y1\nx2 : y2")93        self.assertRunOK("scan --from=x1 --to=z --max_keys=3",94                "x1 : y1\nx2 : y2\nx3 : y3")95        self.assertRunOK("scan --from=x1 --to=z --max_keys=4",96                "x1 : y1\nx2 : y2\nx3 : y3")97        self.assertRunOK("scan --from=x1 --to=x2", "x1 : y1")98        self.assertRunOK("scan --from=x2 --to=x4", "x2 : y2\nx3 : y3")99        self.assertRunFAIL("scan --from=x4 --to=z") # No results => FAIL100        self.assertRunFAIL("scan --from=x1 --to=z --max_keys=foo")101        self.assertRunOK("scan", "x1 : y1\nx2 : y2\nx3 : y3")102        self.assertRunOK("delete x1", "OK")103        self.assertRunOK("scan", "x2 : y2\nx3 : y3")104        self.assertRunOK("delete NonExistentKey", "OK")105        # It is weird that GET and SCAN raise exception for106        # non-existent key, while delete does not107        self.assertRunOK("checkconsistency", "OK")108    def dumpDb(self, params, dumpFile):109        return 0 == run_err_null("./ldb dump %s > %s" % (params, dumpFile))110    def loadDb(self, params, dumpFile):111        return 0 == run_err_null("cat %s | ./ldb load %s" % (dumpFile, params))112    def testStringBatchPut(self):113        print "Running testStringBatchPut..."114        self.assertRunOK("batchput x1 y1 --create_if_missing", "OK")115        self.assertRunOK("scan", "x1 : y1")116        self.assertRunOK("batchput x2 y2 x3 y3 \"x4 abc\" \"y4 xyz\"", "OK")117        self.assertRunOK("scan", "x1 : y1\nx2 : y2\nx3 : y3\nx4 abc : y4 xyz")118        self.assertRunFAIL("batchput")119        self.assertRunFAIL("batchput k1")120        self.assertRunFAIL("batchput k1 v1 k2")121    def testCountDelimDump(self):122        print "Running testCountDelimDump..."123        self.assertRunOK("batchput x.1 x1 --create_if_missing", "OK")124        self.assertRunOK("batchput y.abc abc y.2 2 z.13c pqr", "OK")125        self.assertRunOK("dump --count_delim", "x => count:1\tsize:5\ny => count:2\tsize:12\nz => count:1\tsize:8")126        self.assertRunOK("dump --count_delim=\".\"", "x => count:1\tsize:5\ny => count:2\tsize:12\nz => count:1\tsize:8")127        self.assertRunOK("batchput x,2 x2 x,abc xabc", "OK")128        self.assertRunOK("dump --count_delim=\",\"", "x => count:2\tsize:14\nx.1 => count:1\tsize:5\ny.2 => count:1\tsize:4\ny.abc => count:1\tsize:8\nz.13c => count:1\tsize:8")129    def testCountDelimIDump(self):130        print "Running testCountDelimIDump..."131        self.assertRunOK("batchput x.1 x1 --create_if_missing", "OK")132        self.assertRunOK("batchput y.abc abc y.2 2 z.13c pqr", "OK")133        self.assertRunOK("dump --count_delim", "x => count:1\tsize:5\ny => count:2\tsize:12\nz => count:1\tsize:8")134        self.assertRunOK("dump --count_delim=\".\"", "x => count:1\tsize:5\ny => count:2\tsize:12\nz => count:1\tsize:8")135        self.assertRunOK("batchput x,2 x2 x,abc xabc", "OK")136        self.assertRunOK("dump --count_delim=\",\"", "x => count:2\tsize:14\nx.1 => count:1\tsize:5\ny.2 => count:1\tsize:4\ny.abc => count:1\tsize:8\nz.13c => count:1\tsize:8")137    def testInvalidCmdLines(self):138        print "Running testInvalidCmdLines..."139        # db not specified140        self.assertRunFAILFull("put 0x6133 0x6233 --hex --create_if_missing")141        # No param called he142        self.assertRunFAIL("put 0x6133 0x6233 --he --create_if_missing")143        # max_keys is not applicable for put144        self.assertRunFAIL("put 0x6133 0x6233 --max_keys=1 --create_if_missing")145        # hex has invalid boolean value146    def testHexPutGet(self):147        print "Running testHexPutGet..."148        self.assertRunOK("put a1 b1 --create_if_missing", "OK")149        self.assertRunOK("scan", "a1 : b1")150        self.assertRunOK("scan --hex", "0x6131 : 0x6231")151        self.assertRunFAIL("put --hex 6132 6232")152        self.assertRunOK("put --hex 0x6132 0x6232", "OK")153        self.assertRunOK("scan --hex", "0x6131 : 0x6231\n0x6132 : 0x6232")154        self.assertRunOK("scan", "a1 : b1\na2 : b2")155        self.assertRunOK("get a1", "b1")156        self.assertRunOK("get --hex 0x6131", "0x6231")157        self.assertRunOK("get a2", "b2")158        self.assertRunOK("get --hex 0x6132", "0x6232")159        self.assertRunOK("get --key_hex 0x6132", "b2")160        self.assertRunOK("get --key_hex --value_hex 0x6132", "0x6232")161        self.assertRunOK("get --value_hex a2", "0x6232")162        self.assertRunOK("scan --key_hex --value_hex",163                "0x6131 : 0x6231\n0x6132 : 0x6232")164        self.assertRunOK("scan --hex --from=0x6131 --to=0x6133",165                "0x6131 : 0x6231\n0x6132 : 0x6232")166        self.assertRunOK("scan --hex --from=0x6131 --to=0x6132",167                "0x6131 : 0x6231")168        self.assertRunOK("scan --key_hex", "0x6131 : b1\n0x6132 : b2")169        self.assertRunOK("scan --value_hex", "a1 : 0x6231\na2 : 0x6232")170        self.assertRunOK("batchput --hex 0x6133 0x6233 0x6134 0x6234", "OK")171        self.assertRunOK("scan", "a1 : b1\na2 : b2\na3 : b3\na4 : b4")172        self.assertRunOK("delete --hex 0x6133", "OK")173        self.assertRunOK("scan", "a1 : b1\na2 : b2\na4 : b4")174        self.assertRunOK("checkconsistency", "OK")175    def testTtlPutGet(self):176        print "Running testTtlPutGet..."177        self.assertRunOK("put a1 b1 --ttl --create_if_missing", "OK")178        self.assertRunOK("scan --hex", "0x6131 : 0x6231", True)179        self.assertRunOK("dump --ttl ", "a1 ==> b1", True)180        self.assertRunOK("dump --hex --ttl ",181                         "0x6131 ==> 0x6231\nKeys in range: 1")182        self.assertRunOK("scan --hex --ttl", "0x6131 : 0x6231")183        self.assertRunOK("get --value_hex a1", "0x6231", True)184        self.assertRunOK("get --ttl a1", "b1")185        self.assertRunOK("put a3 b3 --create_if_missing", "OK")186        # fails because timstamp's length is greater than value's187        self.assertRunFAIL("get --ttl a3")188        self.assertRunOK("checkconsistency", "OK")189    def testInvalidCmdLines(self):190        print "Running testInvalidCmdLines..."191        # db not specified192        self.assertRunFAILFull("put 0x6133 0x6233 --hex --create_if_missing")193        # No param called he194        self.assertRunFAIL("put 0x6133 0x6233 --he --create_if_missing")195        # max_keys is not applicable for put196        self.assertRunFAIL("put 0x6133 0x6233 --max_keys=1 --create_if_missing")197        # hex has invalid boolean value198        self.assertRunFAIL("put 0x6133 0x6233 --hex=Boo --create_if_missing")199    def testDumpLoad(self):200        print "Running testDumpLoad..."201        self.assertRunOK("batchput --create_if_missing x1 y1 x2 y2 x3 y3 x4 y4",202                "OK")203        self.assertRunOK("scan", "x1 : y1\nx2 : y2\nx3 : y3\nx4 : y4")204        origDbPath = os.path.join(self.TMP_DIR, self.DB_NAME)205        # Dump and load without any additional params specified206        dumpFilePath = os.path.join(self.TMP_DIR, "dump1")207        loadedDbPath = os.path.join(self.TMP_DIR, "loaded_from_dump1")208        self.assertTrue(self.dumpDb("--db=%s" % origDbPath, dumpFilePath))209        self.assertTrue(self.loadDb(210            "--db=%s --create_if_missing" % loadedDbPath, dumpFilePath))211        self.assertRunOKFull("scan --db=%s" % loadedDbPath,212                "x1 : y1\nx2 : y2\nx3 : y3\nx4 : y4")213        # Dump and load in hex214        dumpFilePath = os.path.join(self.TMP_DIR, "dump2")215        loadedDbPath = os.path.join(self.TMP_DIR, "loaded_from_dump2")216        self.assertTrue(self.dumpDb("--db=%s --hex" % origDbPath, dumpFilePath))217        self.assertTrue(self.loadDb(218            "--db=%s --hex --create_if_missing" % loadedDbPath, dumpFilePath))219        self.assertRunOKFull("scan --db=%s" % loadedDbPath,220                "x1 : y1\nx2 : y2\nx3 : y3\nx4 : y4")221        # Dump only a portion of the key range222        dumpFilePath = os.path.join(self.TMP_DIR, "dump3")223        loadedDbPath = os.path.join(self.TMP_DIR, "loaded_from_dump3")224        self.assertTrue(self.dumpDb(225            "--db=%s --from=x1 --to=x3" % origDbPath, dumpFilePath))226        self.assertTrue(self.loadDb(227            "--db=%s --create_if_missing" % loadedDbPath, dumpFilePath))228        self.assertRunOKFull("scan --db=%s" % loadedDbPath, "x1 : y1\nx2 : y2")229        # Dump upto max_keys rows230        dumpFilePath = os.path.join(self.TMP_DIR, "dump4")231        loadedDbPath = os.path.join(self.TMP_DIR, "loaded_from_dump4")232        self.assertTrue(self.dumpDb(233            "--db=%s --max_keys=3" % origDbPath, dumpFilePath))234        self.assertTrue(self.loadDb(235            "--db=%s --create_if_missing" % loadedDbPath, dumpFilePath))236        self.assertRunOKFull("scan --db=%s" % loadedDbPath,237                "x1 : y1\nx2 : y2\nx3 : y3")238        # Load into an existing db, create_if_missing is not specified239        self.assertTrue(self.dumpDb("--db=%s" % origDbPath, dumpFilePath))240        self.assertTrue(self.loadDb("--db=%s" % loadedDbPath, dumpFilePath))241        self.assertRunOKFull("scan --db=%s" % loadedDbPath,242                "x1 : y1\nx2 : y2\nx3 : y3\nx4 : y4")243        # Dump and load with WAL disabled244        dumpFilePath = os.path.join(self.TMP_DIR, "dump5")245        loadedDbPath = os.path.join(self.TMP_DIR, "loaded_from_dump5")246        self.assertTrue(self.dumpDb("--db=%s" % origDbPath, dumpFilePath))247        self.assertTrue(self.loadDb(248            "--db=%s --disable_wal --create_if_missing" % loadedDbPath,249            dumpFilePath))250        self.assertRunOKFull("scan --db=%s" % loadedDbPath,251                "x1 : y1\nx2 : y2\nx3 : y3\nx4 : y4")252        # Dump and load with lots of extra params specified253        extraParams = " ".join(["--bloom_bits=14", "--compression_type=bzip2",254                                "--block_size=1024", "--auto_compaction=true",255                                "--write_buffer_size=4194304",256                                "--file_size=2097152"])257        dumpFilePath = os.path.join(self.TMP_DIR, "dump6")258        loadedDbPath = os.path.join(self.TMP_DIR, "loaded_from_dump6")259        self.assertTrue(self.dumpDb(260            "--db=%s %s" % (origDbPath, extraParams), dumpFilePath))261        self.assertTrue(self.loadDb(262            "--db=%s %s --create_if_missing" % (loadedDbPath, extraParams),263            dumpFilePath))264        self.assertRunOKFull("scan --db=%s" % loadedDbPath,265                "x1 : y1\nx2 : y2\nx3 : y3\nx4 : y4")266        # Dump with count_only267        dumpFilePath = os.path.join(self.TMP_DIR, "dump7")268        loadedDbPath = os.path.join(self.TMP_DIR, "loaded_from_dump7")269        self.assertTrue(self.dumpDb(270            "--db=%s --count_only" % origDbPath, dumpFilePath))271        self.assertTrue(self.loadDb(272            "--db=%s --create_if_missing" % loadedDbPath, dumpFilePath))273        # DB should have atleast one value for scan to work274        self.assertRunOKFull("put --db=%s k1 v1" % loadedDbPath, "OK")275        self.assertRunOKFull("scan --db=%s" % loadedDbPath, "k1 : v1")276        # Dump command fails because of typo in params277        dumpFilePath = os.path.join(self.TMP_DIR, "dump8")278        self.assertFalse(self.dumpDb(279            "--db=%s --create_if_missing" % origDbPath, dumpFilePath))280    def testMiscAdminTask(self):281        print "Running testMiscAdminTask..."282        # These tests need to be improved; for example with asserts about283        # whether compaction or level reduction actually took place.284        self.assertRunOK("batchput --create_if_missing x1 y1 x2 y2 x3 y3 x4 y4",285                "OK")286        self.assertRunOK("scan", "x1 : y1\nx2 : y2\nx3 : y3\nx4 : y4")287        origDbPath = os.path.join(self.TMP_DIR, self.DB_NAME)288        self.assertTrue(0 == run_err_null(289            "./ldb compact --db=%s" % origDbPath))290        self.assertRunOK("scan", "x1 : y1\nx2 : y2\nx3 : y3\nx4 : y4")291        self.assertTrue(0 == run_err_null(292            "./ldb reduce_levels --db=%s --new_levels=2" % origDbPath))293        self.assertRunOK("scan", "x1 : y1\nx2 : y2\nx3 : y3\nx4 : y4")294        self.assertTrue(0 == run_err_null(295            "./ldb reduce_levels --db=%s --new_levels=3" % origDbPath))296        self.assertRunOK("scan", "x1 : y1\nx2 : y2\nx3 : y3\nx4 : y4")297        self.assertTrue(0 == run_err_null(298            "./ldb compact --db=%s --from=x1 --to=x3" % origDbPath))299        self.assertRunOK("scan", "x1 : y1\nx2 : y2\nx3 : y3\nx4 : y4")300        self.assertTrue(0 == run_err_null(301            "./ldb compact --db=%s --hex --from=0x6131 --to=0x6134"302            % origDbPath))303        self.assertRunOK("scan", "x1 : y1\nx2 : y2\nx3 : y3\nx4 : y4")304        #TODO(dilip): Not sure what should be passed to WAL.Currently corrupted.305        self.assertTrue(0 == run_err_null(306            "./ldb dump_wal --db=%s --walfile=%s --header" % (307                origDbPath, os.path.join(origDbPath, "LOG"))))308        self.assertRunOK("scan", "x1 : y1\nx2 : y2\nx3 : y3\nx4 : y4")309    def testCheckConsistency(self):310        print "Running testCheckConsistency..."311        dbPath = os.path.join(self.TMP_DIR, self.DB_NAME)312        self.assertRunOK("put x1 y1 --create_if_missing", "OK")313        self.assertRunOK("put x2 y2", "OK")314        self.assertRunOK("get x1", "y1")315        self.assertRunOK("checkconsistency", "OK")316        sstFilePath = my_check_output("ls %s" % os.path.join(dbPath, "*.sst"),317                                      shell=True)318        # Modify the file319        my_check_output("echo 'evil' > %s" % sstFilePath, shell=True)320        self.assertRunFAIL("checkconsistency")321        # Delete the file322        my_check_output("rm -f %s" % sstFilePath, shell=True)323        self.assertRunFAIL("checkconsistency")324if __name__ == "__main__":...

test_query.py

Source:test_query.py

...33        def destroy(self):34            self.destroyed = True35    def test_entry_ok_blank(self):36        dialog = self.Dummy_Query(' ')37        self.assertEqual(dialog.entry_ok(), None)38        self.assertEqual((dialog.result, dialog.destroyed), (None, False))39        self.assertIn('blank line', dialog.entry_error['text'])40    def test_entry_ok_good(self):41        dialog = self.Dummy_Query('  good ')42        Equal = self.assertEqual43        Equal(dialog.entry_ok(), 'good')44        Equal((dialog.result, dialog.destroyed), (None, False))45        Equal(dialog.entry_error['text'], '')46    def test_ok_blank(self):47        dialog = self.Dummy_Query('')48        dialog.entry.focus_set = mock.Mock()49        self.assertEqual(dialog.ok(), None)50        self.assertTrue(dialog.entry.focus_set.called)51        del dialog.entry.focus_set52        self.assertEqual((dialog.result, dialog.destroyed), (None, False))53    def test_ok_good(self):54        dialog = self.Dummy_Query('good')55        self.assertEqual(dialog.ok(), None)56        self.assertEqual((dialog.result, dialog.destroyed), ('good', True))57    def test_cancel(self):58        dialog = self.Dummy_Query('does not matter')59        self.assertEqual(dialog.cancel(), None)60        self.assertEqual((dialog.result, dialog.destroyed), (None, True))61class SectionNameTest(unittest.TestCase):62    "Test SectionName subclass of Query."63    class Dummy_SectionName:64        entry_ok = query.SectionName.entry_ok  # Function being tested.65        used_names = ['used']66        def __init__(self, dummy_entry):67            self.entry = Var(value=dummy_entry)68            self.entry_error = {'text': ''}69        def showerror(self, message):70            self.entry_error['text'] = message71    def test_blank_section_name(self):72        dialog = self.Dummy_SectionName(' ')73        self.assertEqual(dialog.entry_ok(), None)74        self.assertIn('no name', dialog.entry_error['text'])75    def test_used_section_name(self):76        dialog = self.Dummy_SectionName('used')77        self.assertEqual(dialog.entry_ok(), None)78        self.assertIn('use', dialog.entry_error['text'])79    def test_long_section_name(self):80        dialog = self.Dummy_SectionName('good'*8)81        self.assertEqual(dialog.entry_ok(), None)82        self.assertIn('longer than 30', dialog.entry_error['text'])83    def test_good_section_name(self):84        dialog = self.Dummy_SectionName('  good ')85        self.assertEqual(dialog.entry_ok(), 'good')86        self.assertEqual(dialog.entry_error['text'], '')87class ModuleNameTest(unittest.TestCase):88    "Test ModuleName subclass of Query."89    class Dummy_ModuleName:90        entry_ok = query.ModuleName.entry_ok  # Function being tested.91        text0 = ''92        def __init__(self, dummy_entry):93            self.entry = Var(value=dummy_entry)94            self.entry_error = {'text': ''}95        def showerror(self, message):96            self.entry_error['text'] = message97    def test_blank_module_name(self):98        dialog = self.Dummy_ModuleName(' ')99        self.assertEqual(dialog.entry_ok(), None)100        self.assertIn('no name', dialog.entry_error['text'])101    def test_bogus_module_name(self):102        dialog = self.Dummy_ModuleName('__name_xyz123_should_not_exist__')103        self.assertEqual(dialog.entry_ok(), None)104        self.assertIn('not found', dialog.entry_error['text'])105    def test_c_source_name(self):106        dialog = self.Dummy_ModuleName('itertools')107        self.assertEqual(dialog.entry_ok(), None)108        self.assertIn('source-based', dialog.entry_error['text'])109    def test_good_module_name(self):110        dialog = self.Dummy_ModuleName('idlelib')111        self.assertTrue(dialog.entry_ok().endswith('__init__.py'))112        self.assertEqual(dialog.entry_error['text'], '')113class GotoTest(unittest.TestCase):114    "Test Goto subclass of Query."115    class Dummy_ModuleName:116        entry_ok = query.Goto.entry_ok  # Function being tested.117        def __init__(self, dummy_entry):118            self.entry = Var(value=dummy_entry)119            self.entry_error = {'text': ''}120        def showerror(self, message):121            self.entry_error['text'] = message122    def test_bogus_goto(self):123        dialog = self.Dummy_ModuleName('a')124        self.assertEqual(dialog.entry_ok(), None)125        self.assertIn('not a base 10 integer', dialog.entry_error['text'])126    def test_bad_goto(self):127        dialog = self.Dummy_ModuleName('0')128        self.assertEqual(dialog.entry_ok(), None)129        self.assertIn('not a positive integer', dialog.entry_error['text'])130    def test_good_goto(self):131        dialog = self.Dummy_ModuleName('1')132        self.assertEqual(dialog.entry_ok(), 1)133        self.assertEqual(dialog.entry_error['text'], '')134# 3 HelpSource test classes each test one method.135class HelpsourceBrowsefileTest(unittest.TestCase):136    "Test browse_file method of ModuleName subclass of Query."137    class Dummy_HelpSource:138        browse_file = query.HelpSource.browse_file139        pathvar = Var()140    def test_file_replaces_path(self):141        dialog = self.Dummy_HelpSource()142        # Path is widget entry, either '' or something.143        # Func return is file dialog return, either '' or something.144        # Func return should override widget entry.145        # We need all 4 combinations to test all (most) code paths.146        for path, func, result in (147                ('', lambda a,b,c:'', ''),148                ('', lambda a,b,c: __file__, __file__),149                ('htest', lambda a,b,c:'', 'htest'),150                ('htest', lambda a,b,c: __file__, __file__)):151            with self.subTest():152                dialog.pathvar.set(path)153                dialog.askfilename = func154                dialog.browse_file()155                self.assertEqual(dialog.pathvar.get(), result)156class HelpsourcePathokTest(unittest.TestCase):157    "Test path_ok method of HelpSource subclass of Query."158    class Dummy_HelpSource:159        path_ok = query.HelpSource.path_ok160        def __init__(self, dummy_path):161            self.path = Var(value=dummy_path)162            self.path_error = {'text': ''}163        def showerror(self, message, widget=None):164            self.path_error['text'] = message165    orig_platform = query.platform  # Set in test_path_ok_file.166    @classmethod167    def tearDownClass(cls):168        query.platform = cls.orig_platform169    def test_path_ok_blank(self):170        dialog = self.Dummy_HelpSource(' ')171        self.assertEqual(dialog.path_ok(), None)172        self.assertIn('no help file', dialog.path_error['text'])173    def test_path_ok_bad(self):174        dialog = self.Dummy_HelpSource(__file__ + 'bad-bad-bad')175        self.assertEqual(dialog.path_ok(), None)176        self.assertIn('not exist', dialog.path_error['text'])177    def test_path_ok_web(self):178        dialog = self.Dummy_HelpSource('')179        Equal = self.assertEqual180        for url in 'www.py.org', 'http://py.org':181            with self.subTest():182                dialog.path.set(url)183                self.assertEqual(dialog.path_ok(), url)184                self.assertEqual(dialog.path_error['text'], '')185    def test_path_ok_file(self):186        dialog = self.Dummy_HelpSource('')187        for platform, prefix in ('darwin', 'file://'), ('other', ''):188            with self.subTest():189                query.platform = platform190                dialog.path.set(__file__)191                self.assertEqual(dialog.path_ok(), prefix + __file__)192                self.assertEqual(dialog.path_error['text'], '')193class HelpsourceEntryokTest(unittest.TestCase):194    "Test entry_ok method of HelpSource subclass of Query."195    class Dummy_HelpSource:196        entry_ok = query.HelpSource.entry_ok197        entry_error = {}198        path_error = {}199        def item_ok(self):200            return self.name201        def path_ok(self):202            return self.path203    def test_entry_ok_helpsource(self):204        dialog = self.Dummy_HelpSource()205        for name, path, result in ((None, None, None),206                                   (None, 'doc.txt', None),207                                   ('doc', None, None),208                                   ('doc', 'doc.txt', ('doc', 'doc.txt'))):209            with self.subTest():210                dialog.name, dialog.path = name, path211                self.assertEqual(dialog.entry_ok(), result)212# 2 CustomRun test classes each test one method.213class CustomRunCLIargsokTest(unittest.TestCase):214    "Test cli_ok method of the CustomRun subclass of Query."215    class Dummy_CustomRun:216        cli_args_ok = query.CustomRun.cli_args_ok217        def __init__(self, dummy_entry):218            self.entry = Var(value=dummy_entry)219            self.entry_error = {'text': ''}220        def showerror(self, message):221            self.entry_error['text'] = message222    def test_blank_args(self):223        dialog = self.Dummy_CustomRun(' ')224        self.assertEqual(dialog.cli_args_ok(), [])225    def test_invalid_args(self):226        dialog = self.Dummy_CustomRun("'no-closing-quote")227        self.assertEqual(dialog.cli_args_ok(), None)228        self.assertIn('No closing', dialog.entry_error['text'])229    def test_good_args(self):230        args = ['-n', '10', '--verbose', '-p', '/path', '--name']231        dialog = self.Dummy_CustomRun(' '.join(args) + ' "my name"')232        self.assertEqual(dialog.cli_args_ok(), args + ["my name"])233        self.assertEqual(dialog.entry_error['text'], '')234class CustomRunEntryokTest(unittest.TestCase):235    "Test entry_ok method of the CustomRun subclass of Query."236    class Dummy_CustomRun:237        entry_ok = query.CustomRun.entry_ok238        entry_error = {}239        restartvar = Var()240        def cli_args_ok(self):241            return self.cli_args242    def test_entry_ok_customrun(self):243        dialog = self.Dummy_CustomRun()244        for restart in {True, False}:245            dialog.restartvar.set(restart)246            for cli_args, result in ((None, None),247                                     (['my arg'], (['my arg'], restart))):248                with self.subTest(restart=restart, cli_args=cli_args):249                    dialog.cli_args = cli_args250                    self.assertEqual(dialog.entry_ok(), result)251# GUI TESTS252class QueryGuiTest(unittest.TestCase):253    @classmethod254    def setUpClass(cls):255        requires('gui')256        cls.root = root = Tk()257        cls.root.withdraw()258        cls.dialog = query.Query(root, 'TEST', 'test', _utest=True)259        cls.dialog.destroy = mock.Mock()260    @classmethod261    def tearDownClass(cls):262        del cls.dialog.destroy263        del cls.dialog264        cls.root.destroy()265        del cls.root266    def setUp(self):267        self.dialog.entry.delete(0, 'end')268        self.dialog.result = None269        self.dialog.destroy.reset_mock()270    def test_click_ok(self):271        dialog = self.dialog272        dialog.entry.insert(0, 'abc')273        dialog.button_ok.invoke()274        self.assertEqual(dialog.result, 'abc')275        self.assertTrue(dialog.destroy.called)276    def test_click_blank(self):277        dialog = self.dialog278        dialog.button_ok.invoke()279        self.assertEqual(dialog.result, None)280        self.assertFalse(dialog.destroy.called)281    def test_click_cancel(self):282        dialog = self.dialog283        dialog.entry.insert(0, 'abc')284        dialog.button_cancel.invoke()...

views.py

Source:views.py

1from django.shortcuts import render2from django.http import HttpResponse3from django.http import JsonResponse4from datetime import datetime5from django.utils.dateparse import parse_datetime6import json7import requests8from kafka import SimpleProducer, KafkaClient9from elasticsearch import Elasticsearch10# Create your views here.11kafka = KafkaClient('kafka:9092')12producer = SimpleProducer(kafka)13es = Elasticsearch(['es'])14def ride_detail(request, ride):15    if request.method != 'GET':16        return JsonResponse({'ok': False, 'error': 'Wrong request type, should be GET'})17    r = requests.get('http://models-api:8000/models/get_ride/' + ride)18    ok = json.loads(r.text)['ok']19    if(ok != True):20        return JsonResponse({'ok':False})21    ride = json.loads(r.text)['car']22    details = json.loads(ride[1:-1])['fields']23    driver_pk = details['driver']24    vehicle_pk = details['car']25    r2 = requests.get('http://models-api:8000/models/get_user/' + str(driver_pk))26    ok = json.loads(r2.text)['ok']27    if(ok != True):28        driver = "Driver Not Found"29    else:30        driver_model = json.loads(r2.text)['user']31        driver_details = json.loads(driver_model)['fields']32        driver = driver_details['first'] + " " + driver_details['last']33    r3 = requests.get('http://models-api:8000/models/get_car/' + str(vehicle_pk))34    ok = json.loads(r3.text)['ok']35    if(ok != True):36        vMake = "Not Found"37        vModel = "Not Found"38    else:39        car_model = json.loads(r3.text)['car']40        car_details = json.loads(car_model)['fields']41        vMake = car_details['make']42        vModel = car_details['model']43    return JsonResponse({'ok':True, 'driver': driver, 'vMake': vMake, 'vModel': vModel, 'leave': details['leave_time'], 'start': details['start'], 'arrive': details['arrive_time'], 'Destination': details['destination']})44def home_detail(request):45    if request.method != 'GET':46        return JsonResponse({'ok': False, 'error': 'Wrong request type, should be GET'})47    r = requests.get('http://models-api:8000/models/all_rides')48    ok = json.loads(r.text)['ok']49    if(ok != True):50        return JsonResponse({'ok':False})51    ride = json.loads(r.text)['ride']52    result_set = []53    for r in ride:54        details = json.loads(r)['fields']55        pk = json.loads(r)['pk']56        driver_pk = details['driver']57        vehicle_pk = details['car']58        req_driver = requests.get('http://models-api:8000/models/get_user/' + str(driver_pk))59        req_vehicle = requests.get('http://models-api:8000/models/get_car/' + str(vehicle_pk))60        resp_driver = json.loads(json.loads(req_driver.text)['user'])['fields']            61        resp_vehicle = json.loads(json.loads(req_vehicle.text)['car'])['fields']62        leavetime = parse_datetime(details['leave_time'])63        arrivetime = parse_datetime(details['arrive_time'])64        ride_info = {'ID':pk ,'driver':resp_driver['first'], 'vMake': resp_vehicle['make'], 'vModel':resp_vehicle['model'], 'leave': leavetime.strftime("%B %d %-I:%M:%S %p"), 'start': details['start'], 'arrive': arrivetime.strftime("%B %d %-I:%M:%S %p"), 'Destination': details['destination']}65        result_set.append(ride_info)66    return JsonResponse({'ok':True, 'result_set': result_set})67def create_user(request):68    if request.method != 'POST':69        return JsonResponse({'ok': False, 'error': 'Wrong request type, should be POST'})70    r = requests.post('http://models-api:8000/models/add_user', data=request.POST)71    ok = json.loads(r.text)['ok']72    if ok:73        userpass = {}74        userpass['username'] = request.POST['username']75        userpass['password'] = request.POST['password']76        r2 = requests.post('http://models-api:8000/models/get_auth', data=userpass)77        d2 = json.loads(r2.text)['ok']78        if d2:79            auth = json.loads(r2.text)['auth']80        else:81            return JsonResponse({'ok': False, 'error': 'Encountered error while authenticating'})82    else:83        return JsonResponse({'ok': False, 'error': 'Encountered error while creating user'})84    return JsonResponse({'ok': True, 'auth': auth, 'log': 'User Created'})85def login(request):86    if request.method != 'POST':87        return JsonResponse({'ok': False, 'error': 'Wrong request type, should be POST'})88    r = requests.post('http://models-api:8000/models/get_auth', data=request.POST)89    ok = json.loads(r.text)['ok']90    if ok:91        return JsonResponse({'ok': True, 'auth': json.loads(r.text)['auth']})92    else:93        return JsonResponse({'ok': False, 'error': 'Encountered error while authenticating'})94def logout(request):95    if request.method != 'POST':96        return JsonResponse({'ok': False, 'error': 'Wrong request type, should be POST'})97    r = requests.post('http://models-api:8000/models/revoke_auth', data=request.POST)98    ok = json.loads(r.text)['ok']99    if ok:100        return JsonResponse({'ok': True, 'log': 'Authenticator revoked'})101    else:102        return JsonResponse({'ok': False, 'error': 'Encountered error while removing auth'})103def add_new_ride(request):104    if request.method != 'POST':105        return JsonResponse({'ok': False, 'error': 'Wrong request type, should be POST'})106    auth = request.POST['auth']107    r = requests.post('http://models-api:8000/models/is_auth', data={'auth':auth})108    ok = json.loads(r.text)['ok']109    if ok:110        username = json.loads(r.text)['username']111        post_values = request.POST.copy()112        post_values['username'] = username113        r2 = requests.post('http://models-api:8000/models/add_ride', data=post_values)114        d2 = json.loads(r2.text)['ok']115        ride_id = json.loads(r2.text)['id']116        post_copy = request.POST.copy()117        post_copy['id'] = ride_id118        if d2:119            try:120                producer.send_messages(b'new-listings-topic', json.dumps(post_copy).encode('utf-8'))121            except Exception:122                #This is ugly, but if the topic doesn't exist we just try again. More than once is a problem though so we let that happen123                producer.send_messages(b'new-listings-topic', json.dumps(post_copy).encode('utf-8'))124            return JsonResponse({'ok': True, 'log': 'Created Ride'})125        else:126            return JsonResponse({'ok': False, 'error': 'Failed to create ride'})127    else:128        return JsonResponse({'ok': False, 'error': 'Invalid authentication to make ride'})129def add_new_vehicle(request):130    if request.method != 'POST':131        return JsonResponse({'ok': False, 'error': 'Wrong request type, should be POST'})132    auth = request.POST['auth']133    auth_req = requests.post('http://models-api:8000/models/is_auth', data={'auth':auth})134    ok = json.loads(auth_req.text)['ok']135    if ok:136        username = json.loads(auth_req.text)['username']137        post_values = request.POST.copy()138        post_values['username'] = username139        resp = requests.post('http://models-api:8000/models/add_vehicle', data=post_values)140        created = json.loads(resp.text)['ok']141        if created:142            return JsonResponse({'ok': True, 'log': 'Added vehicle'})143        else:144            return JsonResponse({'ok': False, 'error': 'Failed to create vehicle'})145    else:146        return JsonResponse({'ok': False, 'error': 'Invalid authentication to make vehicle'})147def search_result(request):148    if request.method != 'POST':149        return JsonResponse({'ok': False, 'error': 'Wrong request type, should be POST'})150    if 'query' not in request.POST:151        return JsonResponse({'ok': False, 'error': 'No query field'})152    results = es.search(index='listing_index', body={'query':{'query_string':{'query': request.POST['query']}}, 'size':10})...

cdif2rsf.py

Source:cdif2rsf.py

1# !/usr/bin/python2import os3from os import EX_OK, EX_USAGE, EX_UNAVAILABLE, EX_IOERR4import sys5##6# $Rev: 1155 $:     Revision of last commit7# $Author: bdubois $:  Author of last commit8# $Date: 2007-05-03 07:51:55 +0200 (Thu, 03 May 2007) $:    Date of last commit9##10def convertToRsf(cdif_input_file_name, rsf_output_file_name):11	print "Reading from file", cdif_input_file_name + "..."12	13	print "\tFiles...",14	return_code = os.system("python ./writeFiles.py " + cdif_input_file_name)15	if return_code == EX_OK:16		print "[ok]"17	else:18		print "[failed]"19		return EX_UNAVAILABLE20	21	print "\tIncludes...",22	return_code = os.system("python ./writeIncludes.py " + cdif_input_file_name)23	if return_code == EX_OK:24		print "[ok]"25	else:26		print "[failed]"27		return EX_UNAVAILABLE28		29	print "\tConditional Compilation...",30	return_code = os.system("python ./writeCondComp.py " + cdif_input_file_name)31	if return_code == EX_OK:32		print "[ok]"33	else:34		print "[failed]"35		return EX_UNAVAILABLE36	37	print "\tClasses...",38	return_code = os.system("python ./writeClasses.py " + cdif_input_file_name)39	if return_code == EX_OK:40		print "[ok]"41	else:42		print "[failed]"43		return EX_UNAVAILABLE44	45	#print "\tTypeDefs...",46	#return_code = os.system("python ./writeTypedefs.py " + cdif_input_file_name)47	#if return_code == EX_OK:48	#	print "[ok]"49	#else:50		print "[failed]"51		return EX_UNAVAILABLE52	53	print "\tInheritance...",54	return_code = os.system("python ./writeInheritance.py " + cdif_input_file_name)55	if return_code == EX_OK:56		print "[ok]"57	else:58		print "[failed]"59		return EX_UNAVAILABLE60	61	print "\tMethods...",62	return_code = os.system("python ./writeMethods.py " + cdif_input_file_name)63	if return_code == EX_OK:64		print "[ok]"65	else:66		print "[failed]"67		return EX_UNAVAILABLE68	69	print "\tAttributes...",70	return_code = os.system("python ./writeAttributes.py " + cdif_input_file_name)71	if return_code == EX_OK:72		print "[ok]"73	else:74		print "[failed]"75		return EX_UNAVAILABLE76	77	print "\tFunctions...",78	return_code = os.system("python ./writeFunctions.py " + cdif_input_file_name)79	if return_code == EX_OK:80		print "[ok]"81	else:82		print "[failed]"83		return EX_UNAVAILABLE84	85	print "\tGlobalVariables...",86	return_code = os.system("python ./writeGlobalVar.py " + cdif_input_file_name)87	if return_code == EX_OK:88		print "[ok]"89	else:90		print "[failed]"91		return EX_UNAVAILABLE92	93	print "\tAccesses...",94	return_code = os.system("python ./writeAccesses.py " + cdif_input_file_name)95	if return_code == EX_OK:96		print "[ok]"97	else:98		print "[failed]"99		return EX_UNAVAILABLE100	101	print "\tInvocations...",102	return_code = os.system("python ./writeInvocations.py " + cdif_input_file_name)103	if return_code == EX_OK:104		print "[ok]"105	else:106		print "[failed]"107		return EX_UNAVAILABLE108		109	print "\tAnnotations...",110	return_code = os.system("python ./writeAnnotations.py " + cdif_input_file_name)111	if return_code == EX_OK:112		print "[ok]"113	else:114		print "[failed]"115		return EX_UNAVAILABLE116	117	print "\tSize and Complexity Metrics...",118	return_code = os.system("python ./writeMetrics.py " + cdif_input_file_name)119	if return_code == EX_OK:120		print "[ok]"121	else:122		print "[failed]"123		return EX_UNAVAILABLE124		125	print "\tChange Frequency Metrics...",126	return_code = os.system("python ./writeCFMetrics.py " + cdif_input_file_name)127	if return_code == EX_OK:128		print "[ok]"129	else:130		print "[failed]"131		return EX_UNAVAILABLE132	133	print "Merging output...",134	return_code = os.system("python ./makeRsf.py " + rsf_output_file_name)135	if return_code == EX_OK:136		print "[ok]"137	else:138		print "[failed]"139		return EX_UNAVAILABLE140	141	remove_command = "rm -f 	accessesWithIDs.txt attributeBelongsToClass.txt \142								attributeHasClassAsType.txt attributesWithIDs.txt \143								classBelongsToFile.txt classesWithIDs.txt \144								conditionalCompilationBlocks.txt typedefsWithIDs.txt\145								fileBelongsToModule.txt filesWithIDs.txt \146								invokableEntityBelongsToFile.txt functionsWithIDs.txt \147								globalVarsWithIDs.txt globalVarHasClassAsType.txt \148								accessibleEntityBelongsToFile.txt \149								inheritanceWithIDs.txt invocationsWithIDs.txt \150								methodBelongsToClass.txt methodsWithIDs.txt \151								modulesWithIDs.txt metricsWithIDs.txt \152								moduleBelongsToModule.txt includeBelongsToFile.txt \153								methodHasClassAsReturnType.txt functionHasClassAsReturnType.txt \154								entityBelongsToBlock.txt methodVisibility.txt \155								methodSignature.txt attributeSignature.txt attributeVisibility.txt\156								accessesLocations.txt invocationLocations.txt \157								defsWithAssociation.txt cfMetricsWithIDs.txt \158								leftValueAccesses.txt annotations.txt \159								annotationBelongsToEntity.txt"160	161	return_code = os.system(remove_command)162	if return_code != EX_OK:163		print "Error cleaning up."164		return EX_IOERR165		166	print "Output written to", rsf_output_file_name167	168	return EX_OK169if __name__ == '__main__':170	if len(sys.argv) < 2:171	  print "Usage:",sys.argv[0],"cdif-input-file-name rsf-output-file-name"172	  sys.exit(EX_USAGE)173	174	input_file=sys.argv[1]175	176	cdif_input_file_name = sys.argv[1]177	rsf_output_file_name = sys.argv[2]178	179	exitCode = convertToRsf(cdif_input_file_name, rsf_output_file_name)180	...

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.