Commits

Alessio Deiana committed 9059abf

support for multiple webserver with different channels + tests for it

  • Participants
  • Parent commits 98e50ab

Comments (0)

Files changed (2)

circuits/web/events.py

 
     _target = "web"
 
+    def __init__(self, *args, **kwargs):
+        if 'target' in kwargs:
+            self._target = kwargs.pop('target')
+        super(WebEvent, self).__init__(*args, **kwargs)
+
+
 class Request(WebEvent):
     """Request(WebEvent) -> Request WebEvent
 
     args: request, response
     """
-
-    success = "request_success", WebEvent._target
-    failure = "request_failure", WebEvent._target
-    filter = "request_filtered", WebEvent._target
-    start = "request_started", WebEvent._target
-    end = "request_completed", WebEvent._target
+    def __init__(self, *args, **kwargs):
+        super(Request, self).__init__(*args, **kwargs)
+        self.success = "request_success", self._target
+        self.failure = "request_failure", self._target
+        self.filter = "request_filtered", self._target
+        self.start = "request_started", self._target
+        self.end = "request_completed", self._target
 
 class Response(WebEvent):
     """Response(WebEvent) -> Response WebEvent
 
     args: request, response
     """
-
-    success = "response_success", WebEvent._target
-    failure = "response_failure", WebEvent._target
-    filter = "response_filtered", WebEvent._target
-    start = "response_started", WebEvent._target
-    end = "response_completed", WebEvent._target
+    def __init__(self, *args, **kwargs):
+        super(Response, self).__init__(*args, **kwargs)
+        self.success = "response_success", self._target
+        self.failure = "response_failure", self._target
+        self.filter = "response_filtered", self._target
+        self.start = "response_started", self._target
+        self.end = "response_completed", self._target
 
 class Stream(WebEvent):
     """Stream(WebEvent) -> Stream WebEvent
 
     args: request, response
     """
-
-    success = "stream_success", WebEvent._target
-    failure = "stream_failure", WebEvent._target
-    filter = "stream_filtered", WebEvent._target
-    start = "stream_started", WebEvent._target
-    end = "stream_completed", WebEvent._target
+    def __init__(self, *args, **kwargs):
+        super(Stream, self).__init__(*args, **kwargs)
+        self.success = "stream_success", self._target
+        self.failure = "stream_failure", self._target
+        self.filter = "stream_filtered", self._target
+        self.start = "stream_started", self._target
+        self.end = "stream_completed", self._target

tests/web/test_core.py

 #!/usr/bin/env python
 
-from circuits.web import Controller
+from circuits.web import Controller, BaseServer
 
 from .helpers import urlencode, urlopen, HTTPError
+import pytest
 
 
 class Root(Controller):
     def test_notfound(self):
         return self.notfound()
 
-def test(webapp):
+def test_simple(webapp):
     f = urlopen(webapp.server.base)
     s = f.read()
     assert s == b"Hello World!"
         assert e.msg == "Not Found"
     else:
         assert False
+
+def test_dual_server(webapp):
+    server = BaseServer(0, channel='web2')
+    server.register(webapp)
+    try:
+        f = urlopen('%s/alt' % server.base)
+        assert False
+    except HTTPError as e:
+        assert e.code == 404
+