← Back to branch summary

~ubuntu-branches/ubuntu/karmic/pypy/karmic

« back to all changes in this revision

Viewing changes to py/test/rsession/testing/test_hostmanage.py

  • Committer: Bazaar Package Importer
  • Author(s): Alexandre Fayolle
  • Date: 2007-04-13 09:33:09 UTC
  • Revision ID: james.westby@ubuntu.com-20070413093309-yoojh4jcoocu2krz
Tags: upstream-1.0.0
Import upstream version 1.0.0

Show diffs side-by-side

added added

removed removed

Lines of Context:
py/test/rsession/testing/test_hostmanage.py
 
1
 
 
2
""" RSync filter test
 
3
"""
 
4
 
 
5
import py
 
6
from py.__.test.rsession.hostmanage import HostRSync, HostInfo, HostManager
 
7
from py.__.test.rsession.hostmanage import sethomedir, gethomedir, getpath_relto_home
 
8
from py.__.test.rsession import repevent
 
9
 
 
10
class DirSetup:
 
11
    def setup_method(self, method):
 
12
        name = "%s.%s" %(self.__class__.__name__, method.func_name)
 
13
        self.tmpdir = py.test.ensuretemp(name)
 
14
        self.source = self.tmpdir.ensure("source", dir=1)
 
15
        self.dest = self.tmpdir.join("dest")
 
16
 
 
17
class TestHostInfo(DirSetup):
 
18
    def _gethostinfo(self, relpath=""):
 
19
        exampledir = self.tmpdir.join("gethostinfo")
 
20
        if relpath:
 
21
            exampledir = exampledir.join(relpath)
 
22
        assert not exampledir.check()
 
23
        hostinfo = HostInfo("localhost:%s" % exampledir)
 
24
        return hostinfo
 
25
 
 
26
    def test_defaultpath(self):
 
27
        x = HostInfo("localhost:")
 
28
        assert x.hostname == "localhost"
 
29
        assert not x.relpath
 
30
 
 
31
    def test_addrel(self):
 
32
        host = HostInfo("localhost:", addrel="whatever")
 
33
        assert host.inplacelocal 
 
34
        assert not host.relpath 
 
35
        host = HostInfo("localhost:/tmp", addrel="base")
 
36
        assert host.relpath == "/tmp/base"
 
37
        host = HostInfo("localhost:tmp", addrel="base2")
 
38
        assert host.relpath == "tmp/base2"
 
39
 
 
40
    def test_path(self):
 
41
        x = HostInfo("localhost:/tmp")
 
42
        assert x.relpath == "/tmp"
 
43
        assert x.hostname == "localhost"
 
44
        assert not x.inplacelocal 
 
45
 
 
46
    def test_hostid(self):
 
47
        x = HostInfo("localhost:")
 
48
        y = HostInfo("localhost:")
 
49
        assert x.hostid != y.hostid 
 
50
        x = HostInfo("localhost:/tmp")
 
51
        y = HostInfo("localhost:")
 
52
        assert x.hostid != y.hostid 
 
53
 
 
54
    def test_non_existing_hosts(self):
 
55
        host = HostInfo("alskdjalsdkjasldkajlsd")
 
56
        py.test.raises((py.process.cmdexec.Error, IOError, EOFError), 
 
57
                       host.initgateway)
 
58
 
 
59
    def test_remote_has_homedir_as_currentdir(self):
 
60
        host = self._gethostinfo()
 
61
        old = py.path.local.get_temproot().chdir()
 
62
        try:
 
63
            host.initgateway()
 
64
            channel = host.gw.remote_exec(py.code.Source(
 
65
                gethomedir, """
 
66
                import os
 
67
                homedir = gethomedir()
 
68
                curdir = os.getcwd()
 
69
                channel.send((curdir, homedir))
 
70
            """))
 
71
            remote_curdir, remote_homedir = channel.receive()
 
72
            assert remote_curdir == remote_homedir 
 
73
        finally:
 
74
            old.chdir()
 
75
 
 
76
    def test_initgateway_localhost_relpath(self):
 
77
        host = HostInfo("localhost:somedir")
 
78
        host.initgateway()
 
79
        assert host.gw
 
80
        try:
 
81
            homedir = py.path.local._gethomedir() 
 
82
            expected = homedir.join("somedir")
 
83
            assert host.gw_remotepath == str(expected)
 
84
        finally:
 
85
            host.gw.exit()
 
86
 
 
87
    def test_initgateway_ssh_and_remotepath(self):
 
88
        option = py.test.config.option
 
89
        if option.sshtarget is None: 
 
90
            py.test.skip("no known ssh target, use -S to set one")
 
91
        host = HostInfo("%s" % (option.sshtarget, ))
 
92
        # this test should be careful to not write/rsync anything
 
93
        # as the remotepath is the default location 
 
94
        # and may be used in the real world 
 
95
        host.initgateway()
 
96
        assert host.gw
 
97
        assert host.gw_remotepath.endswith(host.relpath)
 
98
        channel = host.gw.remote_exec("""
 
99
            import os
 
100
            homedir = os.environ['HOME']
 
101
            relpath = channel.receive()
 
102
            path = os.path.join(homedir, relpath)
 
103
            channel.send(path) 
 
104
        """)
 
105
        channel.send(host.relpath)
 
106
        res = channel.receive()
 
107
        assert res == host.gw_remotepath
 
108
 
 
109
class TestSyncing(DirSetup): 
 
110
    def _gethostinfo(self):
 
111
        hostinfo = HostInfo("localhost:%s" % self.dest)
 
112
        return hostinfo 
 
113
        
 
114
    def test_hrsync_filter(self):
 
115
        self.source.ensure("dir", "file.txt")
 
116
        self.source.ensure(".svn", "entries")
 
117
        self.source.ensure(".somedotfile", "moreentries")
 
118
        self.source.ensure("somedir", "editfile~")
 
119
        syncer = HostRSync(self.source)
 
120
        l = list(self.source.visit(rec=syncer.filter,
 
121
                                   fil=syncer.filter))
 
122
        assert len(l) == 3
 
123
        basenames = [x.basename for x in l]
 
124
        assert 'dir' in basenames
 
125
        assert 'file.txt' in basenames
 
126
        assert 'somedir' in basenames
 
127
 
 
128
    def test_hrsync_localhost_inplace(self):
 
129
        h1 = HostInfo("localhost")
 
130
        events = []
 
131
        rsync = HostRSync(self.source)
 
132
        h1.initgateway()
 
133
        rsync.add_target_host(h1, reporter=events.append)
 
134
        assert events
 
135
        l = [x for x in events 
 
136
                if isinstance(x, repevent.HostRSyncing)]
 
137
        assert len(l) == 1
 
138
        ev = l[0]
 
139
        assert ev.host == h1
 
140
        assert ev.root == ev.remotepath 
 
141
        l = [x for x in events 
 
142
                if isinstance(x, repevent.HostRSyncRootReady)]
 
143
        assert len(l) == 1
 
144
        ev = l[0]
 
145
        assert ev.root == self.source
 
146
        assert ev.host == h1
 
147
 
 
148
    def test_hrsync_one_host(self):
 
149
        h1 = self._gethostinfo()
 
150
        finished = []
 
151
        rsync = HostRSync(self.source)
 
152
        h1.initgateway()
 
153
        rsync.add_target_host(h1)
 
154
        self.source.join("hello.py").write("world")
 
155
        rsync.send()
 
156
        assert self.dest.join("hello.py").check()
 
157
 
 
158
    def test_hrsync_same_host_twice(self):
 
159
        h1 = self._gethostinfo()
 
160
        h2 = self._gethostinfo()
 
161
        finished = []
 
162
        rsync = HostRSync(self.source)
 
163
        l = []
 
164
        h1.initgateway()
 
165
        h2.initgateway()
 
166
        res1 = rsync.add_target_host(h1)
 
167
        assert res1
 
168
        res2 = rsync.add_target_host(h2)
 
169
        assert not res2
 
170
 
 
171
class TestHostManager(DirSetup):
 
172
    def gethostmanager(self, dist_hosts, dist_rsync_roots=None):
 
173
        l = ["dist_hosts = %r" % dist_hosts]
 
174
        if dist_rsync_roots:
 
175
            l.append("dist_rsync_roots = %r" % dist_rsync_roots)
 
176
        self.source.join("conftest.py").write("\n".join(l))
 
177
        config = py.test.config._reparse([self.source])
 
178
        assert config.topdir == self.source
 
179
        hm = HostManager(config)
 
180
        assert hm.hosts
 
181
        return hm
 
182
        
 
183
    def test_hostmanager_custom_hosts(self):
 
184
        config = py.test.config._reparse([self.source])
 
185
        hm = HostManager(config, hosts=[1,2,3])
 
186
        assert hm.hosts == [1,2,3]
 
187
 
 
188
    def test_hostmanager_init_rsync_topdir(self):
 
189
        dir2 = self.source.ensure("dir1", "dir2", dir=1)
 
190
        dir2.ensure("hello")
 
191
        hm = self.gethostmanager(
 
192
            dist_hosts = ["localhost:%s" % self.dest]
 
193
        )
 
194
        assert hm.config.topdir == self.source
 
195
        hm.init_rsync([].append)
 
196
        dest = self.dest.join(self.source.basename)
 
197
        assert dest.join("dir1").check()
 
198
        assert dest.join("dir1", "dir2").check()
 
199
        assert dest.join("dir1", "dir2", 'hello').check()
 
200
 
 
201
    def test_hostmanager_init_rsync_topdir_explicit(self):
 
202
        dir2 = self.source.ensure("dir1", "dir2", dir=1)
 
203
        dir2.ensure("hello")
 
204
        hm = self.gethostmanager(
 
205
            dist_hosts = ["localhost:%s" % self.dest],
 
206
            dist_rsync_roots = [str(self.source)]
 
207
        )
 
208
        assert hm.config.topdir == self.source
 
209
        hm.init_rsync([].append)
 
210
        dest = self.dest.join(self.source.basename)
 
211
        assert dest.join("dir1").check()
 
212
        assert dest.join("dir1", "dir2").check()
 
213
        assert dest.join("dir1", "dir2", 'hello').check()
 
214
 
 
215
    def test_hostmanager_init_rsync_roots(self):
 
216
        dir2 = self.source.ensure("dir1", "dir2", dir=1)
 
217
        self.source.ensure("dir1", "somefile", dir=1)
 
218
        dir2.ensure("hello")
 
219
        self.source.ensure("bogusdir", "file")
 
220
        self.source.join("conftest.py").write(py.code.Source("""
 
221
            dist_rsync_roots = ['dir1/dir2']
 
222
        """))
 
223
        config = py.test.config._reparse([self.source])
 
224
        hm = HostManager(config, 
 
225
                         hosts=[HostInfo("localhost:" + str(self.dest))])
 
226
        events = []
 
227
        hm.init_rsync(reporter=events.append)
 
228
        assert self.dest.join("dir2").check()
 
229
        assert not self.dest.join("dir1").check()
 
230
        assert not self.dest.join("bogus").check()
 
231
 
 
232
    def test_hostmanager_rsync_ignore(self):
 
233
        dir2 = self.source.ensure("dir1", "dir2", dir=1)
 
234
        dir5 = self.source.ensure("dir5", "dir6", "bogus")
 
235
        dirf = self.source.ensure("dir5", "file")
 
236
        dir2.ensure("hello")
 
237
        self.source.join("conftest.py").write(py.code.Source("""
 
238
            dist_rsync_ignore = ['dir1/dir2', 'dir5/dir6']
 
239
        """))
 
240
        config = py.test.config._reparse([self.source])
 
241
        hm = HostManager(config, 
 
242
                         hosts=[HostInfo("localhost:" + str(self.dest))])
 
243
        events = []
 
244
        print events
 
245
        hm.init_rsync(reporter=events.append)
 
246
        assert self.dest.join("dir1").check()
 
247
        assert not self.dest.join("dir1", "dir2").check()
 
248
        assert self.dest.join("dir5","file").check()
 
249
        assert not self.dest.join("dir6").check()
 
250
 
 
251
    def test_hostmanage_optimise_localhost(self):
 
252
        hosts = [HostInfo("localhost") for i in range(3)]
 
253
        config = py.test.config._reparse([self.source])
 
254
        hm = HostManager(config, hosts=hosts)
 
255
        events = []
 
256
        hm.init_rsync(events.append)
 
257
        for host in hosts:
 
258
            assert host.inplacelocal
 
259
            assert host.gw_remotepath is None
 
260
            assert not host.relpath 
 
261
        assert events
 
262
 
 
263
    def XXXtest_ssh_rsync_samehost_twice(self):
 
264
        #XXX we have no easy way to have a temp directory remotely!
 
265
        option = py.test.config.option
 
266
        if option.sshtarget is None: 
 
267
            py.test.skip("no known ssh target, use -S to set one")
 
268
        host1 = HostInfo("%s" % (option.sshtarget, ))
 
269
        host2 = HostInfo("%s" % (option.sshtarget, ))
 
270
        hm = HostManager(config, hosts=[host1, host2])
 
271
        events = []
 
272
        hm.init_rsync(events.append)
 
273
        print events
 
274
        assert 0
 
275
 
 
276
def test_getpath_relto_home():
 
277
    x = getpath_relto_home("hello")
 
278
    assert x == py.path.local._gethomedir().join("hello")
 
279
    x = getpath_relto_home(".")
 
280
    assert x == py.path.local._gethomedir()
 
281
 
 
282
def test_sethomedir():
 
283
    old = py.path.local.get_temproot().chdir()
 
284
    try:
 
285
        sethomedir()
 
286
        curdir = py.path.local()
 
287
    finally:
 
288
        old.chdir()
 
289
 
 
290
    assert py.path.local._gethomedir() == curdir
 
291