[svn r37276] merging of file:///svn/py/branch/config/py/test/rsession/testing/test_rsession.py

revisions 36936 to 37267: ------------------------------------------------------------------------ r37262 | hpk | 2007-01-24 14:55:00 +0100 (Wed, 24 Jan 2007) | 2 lines fix failing test ------------------------------------------------------------------------ r37015 | fijal | 2007-01-19 17:24:32 +0100 (Fri, 19 Jan 2007) | 2 lines Major cleanup of hostmanage. Got rid of global functions in favor of some better-structured classes. ------------------------------------------------------------------------ r36974 | fijal | 2007-01-18 22:32:04 +0100 (Thu, 18 Jan 2007) | 2 lines Fix the test to actually perform RSync of pylib itself and a test if that was true. ------------------------------------------------------------------------ r36970 | hpk | 2007-01-18 20:33:16 +0100 (Thu, 18 Jan 2007) | 6 lines some little cleanups - getting rid of we_are_remote which is not used anywhere as far as i can see. maciej: please check that i didn't mess anything up, but for me the same tests as before pass ------------------------------------------------------------------------ r36968 | hpk | 2007-01-18 20:00:21 +0100 (Thu, 18 Jan 2007) | 2 lines fixing some imports ------------------------------------------------------------------------ r36966 | fijal | 2007-01-18 19:36:02 +0100 (Thu, 18 Jan 2007) | 2 lines Add try: finally: ------------------------------------------------------------------------ r36965 | fijal | 2007-01-18 19:35:00 +0100 (Thu, 18 Jan 2007) | 2 lines Fix the test. ------------------------------------------------------------------------ r36963 | fijal | 2007-01-18 19:21:43 +0100 (Thu, 18 Jan 2007) | 2 lines Fix remaining tests and make strange hack with parse/reparse. ------------------------------------------------------------------------ r36962 | fijal | 2007-01-18 19:06:13 +0100 (Thu, 18 Jan 2007) | 3 lines Huge refactoring. Got rid of remote_options and session_options, two tests still fail, so it's intermediate checkin. ------------------------------------------------------------------------ r36943 | fijal | 2007-01-18 16:09:30 +0100 (Thu, 18 Jan 2007) | 3 lines at least LSession should work now. Did not touched yet the remote options stuff, but some simplifications were done. ------------------------------------------------------------------------ r36937 | fijal | 2007-01-18 14:22:54 +0100 (Thu, 18 Jan 2007) | 2 lines Create a branch for further config cleanups. ------------------------------------------------------------------------ --HG-- branch : trunk
2007-01-24 17:40:04 +01:00 · 2007-01-24 17:40:04 +01:00 · 86ac16b937
parent 2bdeb13069
commit 86ac16b937
1 changed files with 324 additions and 0 deletions
--- a/py/test/rsession/testing/test_rsession.py.merge.tmp
+++ b/py/test/rsession/testing/test_rsession.py.merge.tmp
@ -0,0 +1,324 @@
+
+""" Tests various aspects of rsession, like ssh hosts setup/teardown
+"""
+
+import py
+from py.__.test.rsession import report
+from py.__.test.rsession.rsession import RSession, parse_directories,\
+    parse_directories
+from py.__.test.rsession.hostmanage import HostOptions, HostManager,\
+     HostInfo
+from py.__.test.rsession.testing.test_slave import funcfail_spec,\
+    funcpass_spec, funcskip_spec, funcprint_spec, funcprintfail_spec, \
+    funcoptioncustom_spec
+
+def setup_module(mod):
+    mod.pkgdir = py.path.local(py.__file__).dirpath()
+
+def test_setup_non_existing_hosts(): 
+    setup_events = []
+    hosts = [HostInfo("alskdjalsdkjasldkajlsd")]
+    hm = HostManager(hosts, None, pkgdir)
+    cmd = "hm.init_hosts(setup_events.append)"
+    py.test.raises((py.process.cmdexec.Error, IOError, EOFError), cmd)
+    #assert setup_events
+
+def test_getpkdir():
+    one = pkgdir.join("initpkg.py")
+    two = pkgdir.join("path", "__init__.py")
+    p1 = RSession.getpkgdir(one)
+    p2 = RSession.getpkgdir(two) 
+    assert p1 == p2
+    assert p1 == pkgdir 
+
+def test_getpkdir_no_inits():
+    tmp = py.test.ensuretemp("getpkdir1")
+    fn = tmp.ensure("hello.py")
+    assert RSession.getpkgdir(fn) == fn
+
+def test_make_colitems():
+    one = pkgdir.join("initpkg.py")
+    two = pkgdir.join("path", "__init__.py")
+
+    cols = RSession.make_colitems([one, two], baseon=pkgdir) 
+    assert len(cols) == 2
+    col_one, col_two = cols
+    assert col_one.listnames() == ["py", "initpkg.py"]
+    assert col_two.listnames() == ["py", "path", "__init__.py"]
+
+    cols = RSession.make_colitems([one, two], baseon=pkgdir.dirpath()) 
+    assert len(cols) == 2
+    col_one, col_two = cols
+    assert col_one.listnames() == [pkgdir.dirpath().basename, 
+                                   "py", "initpkg.py"]
+    assert col_two.listnames() == [pkgdir.dirpath().basename, 
+                                   "py", "path", "__init__.py"]
+
+def test_example_tryiter():
+    events = []
+    tmpdir = py.test.ensuretemp("tryitertest")
+    tmpdir.ensure("a", "__init__.py")
+    tmpdir.ensure("conftest.py").write(py.code.Source("""
+        import py
+        py.test.skip("Reason")
+    """))
+    tmpdir.ensure("a", "test_empty.py").write(py.code.Source("""
+        def test_empty():
+            pass
+    """))
+    rootcol = py.test.collect.Directory(tmpdir)
+    data = list(rootcol.tryiter(reporterror=events.append))
+    assert len(events) == 2
+    assert str(events[1][0].value) == "Reason"
+
+class TestRSessionRemote: 
+    def test_example_distribution_minus_x(self):
+        tmpdir = py.test.ensuretemp("example_distribution_minus_x")
+        tmpdir.ensure("sub", "conftest.py").write(py.code.Source("""
+            disthosts = [%r]
+        """ % ('localhost',)))
+        tmpdir.ensure("sub", "__init__.py")
+        tmpdir.ensure("sub", "test_one.py").write(py.code.Source("""
+            def test_1(): 
+                pass
+            def test_x():
+                import py
+                py.test.skip("aaa")
+            def test_2():
+                assert 0
+            def test_3():
+                raise ValueError(23)
+            def test_4(someargs):
+                pass
+        """))
+        args = [str(tmpdir.join("sub")), "-x"]
+        config = py.test.config._reparse(args)
+        rsession = RSession(config)
+        allevents = []
+        rsession.main(reporter=allevents.append)
+        testevents = [x for x in allevents 
+                        if isinstance(x, report.ReceivedItemOutcome)]
+        assert len(testevents) == 3
+        assert rsession.checkfun()
+
+    def test_example_distribution(self):
+        subdir = "sub_example_dist"
+        tmpdir = py.test.ensuretemp("example_distribution")
+        tmpdir.ensure(subdir, "conftest.py").write(py.code.Source("""
+            disthosts = [%r]
+            distrsync_roots = ["%s", "py"]
+        """ % ('localhost', subdir)))
+        tmpdir.ensure(subdir, "__init__.py")
+        tmpdir.ensure(subdir, "test_one.py").write(py.code.Source("""
+            def test_1(): 
+                pass
+            def test_2():
+                assert 0
+            def test_3():
+                raise ValueError(23)
+            def test_4(someargs):
+                pass
+            def test_5():
+                assert __file__ != '%s'
+            def test_6():
+                import py
+                assert py.__file__ != '%s'
+        """ % (str(tmpdir.join(subdir)), py.__file__)))
+        tmpdir.join("py").mksymlinkto(py.path.local(py.__file__).dirpath())
+        args = [str(tmpdir.join(subdir))]
+        config = py.test.config._reparse(args)
+        rsession = RSession(config, optimise_localhost=False)
+        allevents = []
+        rsession.main(reporter=allevents.append) 
+        testevents = [x for x in allevents 
+                        if isinstance(x, report.ReceivedItemOutcome)]
+        assert len(testevents)
+        passevents = [i for i in testevents if i.outcome.passed]
+        failevents = [i for i in testevents if i.outcome.excinfo]
+        skippedevents = [i for i in testevents if i.outcome.skipped]
+        assert len(testevents) == 6
+        assert len(passevents) == 3
+        assert len(failevents) == 3
+        tb = failevents[0].outcome.excinfo.traceback
+        assert str(tb[0].path).find("test_one") != -1
+        assert tb[0].source.find("test_2") != -1
+        assert failevents[0].outcome.excinfo.typename == 'AssertionError'
+        tb = failevents[1].outcome.excinfo.traceback
+        assert str(tb[0].path).find("test_one") != -1
+        assert tb[0].source.find("test_3") != -1
+        assert failevents[1].outcome.excinfo.typename == 'ValueError'
+        assert failevents[1].outcome.excinfo.value == '23'
+        tb = failevents[2].outcome.excinfo.traceback
+        assert failevents[2].outcome.excinfo.typename == 'TypeError'
+        assert str(tb[0].path).find("executor") != -1
+        assert tb[0].source.find("execute") != -1
+        
+    def test_setup_teardown_ssh(self):
+        hosts = [HostInfo('localhost')]
+        parse_directories(hosts)
+        setup_events = []
+        teardown_events = []
+        
+        config = py.test.config._reparse([])
+        opts = HostOptions(optimise_localhost=False, rsync_roots=['py'])
+        hm = HostManager(hosts, config, pkgdir, opts)
+        nodes = hm.init_hosts(setup_events.append)
+        hm.teardown_hosts(teardown_events.append, 
+                       [node.channel for node in nodes], nodes)
+        
+        count_rsyn_calls = [i for i in setup_events 
+                if isinstance(i, report.HostRSyncing)]
+        assert len(count_rsyn_calls) == len([i for i in hosts])
+        count_ready_calls = [i for i in setup_events 
+                if isinstance(i, report.HostReady)]
+        assert len(count_ready_calls) == len([i for i in hosts])
+        
+        # same for teardown events
+        teardown_wait_starts = [i for i in teardown_events 
+                                    if isinstance(i, report.CallStart)]
+        teardown_wait_ends = [i for i in teardown_events 
+                                    if isinstance(i, report.CallFinish)]
+        assert len(teardown_wait_starts) == len(hosts)
+        assert len(teardown_wait_ends) == len(hosts)
+
+    def test_setup_teardown_run_ssh(self):
+        hosts = [HostInfo('localhost')]
+        parse_directories(hosts)
+        allevents = []
+        
+        config = py.test.config._reparse([])
+        opts = HostOptions(optimise_localhost=False, rsync_roots=['py'])
+        hm = HostManager(hosts, config, pkgdir, opts)
+        nodes = hm.init_hosts(allevents.append)
+        
+        from py.__.test.rsession.testing.test_executor \
+            import ItemTestPassing, ItemTestFailing, ItemTestSkipping
+        
+        rootcol = py.test.collect.Directory(pkgdir.dirpath())
+        itempass = rootcol.getitembynames(funcpass_spec)
+        itemfail = rootcol.getitembynames(funcfail_spec)
+        itemskip = rootcol.getitembynames(funcskip_spec)
+        itemprint = rootcol.getitembynames(funcprint_spec)
+
+        # actually run some tests
+        for node in nodes:
+            node.send(itempass)
+            node.send(itemfail)
+            node.send(itemskip)
+            node.send(itemprint)
+
+        hm.teardown_hosts(allevents.append, [node.channel for node in nodes], nodes)
+
+        events = [i for i in allevents 
+                        if isinstance(i, report.ReceivedItemOutcome)]
+        passed = [i for i in events 
+                        if i.outcome.passed]
+        skipped = [i for i in events 
+                        if i.outcome.skipped]
+        assert len(passed) == 2 * len(nodes)
+        assert len(skipped) == len(nodes)
+        assert len(events) == 4 * len(nodes)
+        # one of passed for each node has non-empty stdout
+        passed_stdout = [i for i in passed if i.outcome.stdout.find('samfing') != -1]
+        assert len(passed_stdout) == len(nodes), passed
+
+    def test_config_pass(self):
+        """ Tests options object passing master -> server
+        """
+        allevents = []
+        hosts = [HostInfo('localhost')]
+        parse_directories(hosts)
+        config = py.test.config._reparse([])
+        config._overwrite('custom', 'custom')
+        # we need to overwrite default list to serialize
+        from py.__.test.rsession.master import defaultconftestnames
+        defaultconftestnames.append("custom")
+        try:
+            opts = HostOptions(optimise_localhost=False, rsync_roots=['py'])
+            hm = HostManager(hosts, config, pkgdir, opts)
+            nodes = hm.init_hosts(allevents.append)
+
+            rootcol = py.test.collect.Directory(pkgdir.dirpath())
+            itempass = rootcol.getitembynames(funcoptioncustom_spec)
+
+            for node in nodes:
+                node.send(itempass)
+
+            hm.teardown_hosts(allevents.append, [node.channel for node in nodes], nodes)
+            events = [i for i in allevents 
+                            if isinstance(i, report.ReceivedItemOutcome)]
+            passed = [i for i in events 
+                            if i.outcome.passed]
+            skipped = [i for i in events 
+                            if i.outcome.skipped]
+            assert len(passed) == 1 * len(nodes)
+            assert len(skipped) == 0
+            assert len(events) == len(passed)
+        finally:
+            defaultconftestnames.remove("custom")
+    
+    def test_nice_level(self):
+        """ Tests if nice level behaviour is ok
+        """
+        allevents = []
+        hosts = [HostInfo('localhost')]
+        parse_directories(hosts)
+        tmpdir = py.test.ensuretemp("nice")
+        tmpdir.ensure("__init__.py")
+        tmpdir.ensure("conftest.py").write(py.code.Source("""
+        disthosts = ['localhost']
+        dist_nicelevel = 10
+        """))
+        tmpdir.ensure("test_one.py").write("""def test_nice():
+            import os
+            assert os.nice(0) == 10
+        """)
+        
+        config = py.test.config._reparse([tmpdir])
+        rsession = RSession(config)
+        allevents = []
+        rsession.main(reporter=allevents.append) 
+        testevents = [x for x in allevents 
+                        if isinstance(x, report.ReceivedItemOutcome)]
+        passevents = [x for x in testevents if x.outcome.passed]
+        assert len(passevents) == 1
+    
+class XxxTestDirectories(object):
+    # need complete rewrite, and unsure if it makes sense at all
+    def test_simple_parse(self):
+        sshhosts = [HostInfo(i) for i in ['h1', 'h2', 'h3']]
+        parse_directories(sshhosts)
+    
+    def test_sophisticated_parse(self):
+        sshhosts = ['a@h1:/tmp', 'h2:tmp', 'h3']
+        dirs = parse_directories(sshhosts)
+        assert py.builtin.sorted(
+            dirs.values()) == ['/tmp', 'pytestcache', 'tmp']
+    
+    def test_parse_multiple_hosts(self):
+        hosts = ['h1', 'h1', 'h1:/tmp']
+        dirs = parse_directories(hosts)
+        assert dirs == {(0, 'h1'): 'pytestcache', (1, 'h1'): 'pytestcache', 
+            (2, 'h1'):'/tmp'}
+
+class TestInithosts(object):
+    def test_inithosts(self):
+        testevents = []
+        hostnames = ['h1:/tmp', 'h1:/tmp', 'h1:/other', 'h2', 'h2:home']
+        hosts = [HostInfo(i) for i in hostnames]
+        parse_directories(hosts)
+        config = py.test.config._reparse([])
+        opts = HostOptions(do_sync=False, create_gateways=False)
+        hm = HostManager(hosts, config, pkgdir, opts)
+        nodes = hm.init_hosts(testevents.append)        
+        events = [i for i in testevents if isinstance(i, report.HostRSyncing)]
+        assert len(events) == 4
+        assert events[0].host.hostname == 'h1'
+        assert events[0].host.relpath == '/tmp-h1'
+        assert events[1].host.hostname == 'h1'
+        assert events[1].host.relpath == '/other-h1'
+        assert events[2].host.hostname == 'h2'
+        assert events[2].host.relpath == 'pytestcache-h2'
+        assert events[3].host.hostname == 'h2'
+        assert events[3].host.relpath == 'home-h2'
+