(evt, numrequests, requestHandler=None, encoding=None)
| 613 | # Second when the server has been shutdown. The user must clear |
| 614 | # the event after it has been set the first time to catch the second set. |
| 615 | def http_server(evt, numrequests, requestHandler=None, encoding=None): |
| 616 | class TestInstanceClass: |
| 617 | def div(self, x, y): |
| 618 | return x // y |
| 619 | |
| 620 | def _methodHelp(self, name): |
| 621 | if name == 'div': |
| 622 | return 'This is the div function' |
| 623 | |
| 624 | class Fixture: |
| 625 | @staticmethod |
| 626 | def getData(): |
| 627 | return '42' |
| 628 | |
| 629 | class MyXMLRPCServer(xmlrpc.server.SimpleXMLRPCServer): |
| 630 | def get_request(self): |
| 631 | # Ensure the socket is always non-blocking. On Linux, socket |
| 632 | # attributes are not inherited like they are on *BSD and Windows. |
| 633 | s, port = self.socket.accept() |
| 634 | s.setblocking(True) |
| 635 | return s, port |
| 636 | |
| 637 | if not requestHandler: |
| 638 | requestHandler = xmlrpc.server.SimpleXMLRPCRequestHandler |
| 639 | serv = MyXMLRPCServer(("localhost", 0), requestHandler, |
| 640 | encoding=encoding, |
| 641 | logRequests=False, bind_and_activate=False) |
| 642 | try: |
| 643 | serv.server_bind() |
| 644 | global ADDR, PORT, URL |
| 645 | ADDR, PORT = serv.socket.getsockname() |
| 646 | #connect to IP address directly. This avoids socket.create_connection() |
| 647 | #trying to connect to "localhost" using all address families, which |
| 648 | #causes slowdown e.g. on vista which supports AF_INET6. The server listens |
| 649 | #on AF_INET only. |
| 650 | URL = "http://%s:%d"%(ADDR, PORT) |
| 651 | serv.server_activate() |
| 652 | serv.register_introspection_functions() |
| 653 | serv.register_multicall_functions() |
| 654 | serv.register_function(pow) |
| 655 | serv.register_function(lambda x: x, 'têšt') |
| 656 | @serv.register_function |
| 657 | def my_function(): |
| 658 | '''This is my function''' |
| 659 | return True |
| 660 | @serv.register_function(name='add') |
| 661 | def _(x, y): |
| 662 | return x + y |
| 663 | testInstance = TestInstanceClass() |
| 664 | serv.register_instance(testInstance, allow_dotted_names=True) |
| 665 | evt.set() |
| 666 | |
| 667 | # handle up to 'numrequests' requests |
| 668 | while numrequests > 0: |
| 669 | serv.handle_request() |
| 670 | numrequests -= 1 |
| 671 | |
| 672 | except TimeoutError: |
no test coverage detected