File: conftest.py

package info (click to toggle)
circuits 3.2.3-3
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 2,980 kB
  • sloc: python: 17,583; javascript: 3,226; makefile: 100
file content (72 lines) | stat: -rw-r--r-- 1,800 bytes parent folder | download | duplicates (2)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
"""py.test config"""

import os

import pytest

from circuits import Component, Debugger, handler
from circuits.net.sockets import close
from circuits.web import Server, Static
from circuits.web.client import Client, request


DOCROOT = os.path.join(os.path.dirname(__file__), 'static')


class WebApp(Component):
    channel = 'web'

    def init(self):
        self.closed = False

        self.server = Server(0).register(self)
        Static('/static', DOCROOT, dirlisting=True).register(self)


class WebClient(Client):
    def init(self, *args, **kwargs):
        self.closed = False

    def __call__(self, method, path, body=None, headers={}):
        waiter = pytest.WaitEvent(self, 'response', channel=self.channel)
        self.fire(request(method, path, body, headers))
        assert waiter.wait()

        return self.response

    @handler('closed', channel='*', priority=1.0)
    def _on_closed(self):
        self.closed = True


@pytest.fixture()
def webapp(request, manager, watcher):
    webapp = WebApp().register(manager)
    assert watcher.wait('ready')

    if hasattr(request.module, 'application'):
        from circuits.web.wsgi import Gateway

        application = request.module.application
        Gateway({'/': application}).register(webapp)
        assert watcher.wait('registered')

    Root = getattr(request.module, 'Root', None)
    if Root is not None:
        Root().register(webapp)
        assert watcher.wait('registered')

    if request.config.option.verbose:
        Debugger().register(webapp)
        assert watcher.wait('registered')

    def finalizer():
        webapp.fire(close())
        assert watcher.wait('closed')

        webapp.unregister()
        assert watcher.wait('unregistered')

    request.addfinalizer(finalizer)

    return webapp