File manager - Edit - /home/newsbmcs.com/public_html/static/img/logo/test_tap.py.tar
Back
usr/lib/python3/dist-packages/twisted/conch/test/test_tap.py 0000644 00000011655 15030056134 0020204 0 ustar 00 # Copyright (c) Twisted Matrix Laboratories. # See LICENSE for details. """ Tests for L{twisted.conch.tap}. """ from twisted.application.internet import StreamServerEndpointService from twisted.cred import error from twisted.cred.credentials import ISSHPrivateKey, IUsernamePassword, UsernamePassword from twisted.python.reflect import requireModule from twisted.trial.unittest import TestCase cryptography = requireModule("cryptography") pyasn1 = requireModule("pyasn1") unix = requireModule("twisted.conch.unix") if cryptography and pyasn1 and unix: from twisted.conch import tap from twisted.conch.openssh_compat.factory import OpenSSHFactory class MakeServiceTests(TestCase): """ Tests for L{tap.makeService}. """ if not cryptography: skip = "can't run without cryptography" if not pyasn1: skip = "Cannot run without PyASN1" if not unix: skip = "can't run on non-posix computers" usernamePassword = (b"iamuser", b"thisispassword") def setUp(self): """ Create a file with two users. """ self.filename = self.mktemp() with open(self.filename, "wb+") as f: f.write(b":".join(self.usernamePassword)) self.options = tap.Options() def test_basic(self): """ L{tap.makeService} returns a L{StreamServerEndpointService} instance running on TCP port 22, and the linked protocol factory is an instance of L{OpenSSHFactory}. """ config = tap.Options() service = tap.makeService(config) self.assertIsInstance(service, StreamServerEndpointService) self.assertEqual(service.endpoint._port, 22) self.assertIsInstance(service.factory, OpenSSHFactory) def test_defaultAuths(self): """ Make sure that if the C{--auth} command-line option is not passed, the default checkers are (for backwards compatibility): SSH and UNIX """ numCheckers = 2 self.assertIn( ISSHPrivateKey, self.options["credInterfaces"], "SSH should be one of the default checkers", ) self.assertIn( IUsernamePassword, self.options["credInterfaces"], "UNIX should be one of the default checkers", ) self.assertEqual( numCheckers, len(self.options["credCheckers"]), "There should be %d checkers by default" % (numCheckers,), ) def test_authAdded(self): """ The C{--auth} command-line option will add a checker to the list of checkers, and it should be the only auth checker """ self.options.parseOptions(["--auth", "file:" + self.filename]) self.assertEqual(len(self.options["credCheckers"]), 1) def test_multipleAuthAdded(self): """ Multiple C{--auth} command-line options will add all checkers specified to the list ofcheckers, and there should only be the specified auth checkers (no default checkers). """ self.options.parseOptions( [ "--auth", "file:" + self.filename, "--auth", "memory:testuser:testpassword", ] ) self.assertEqual(len(self.options["credCheckers"]), 2) def test_authFailure(self): """ The checker created by the C{--auth} command-line option returns a L{Deferred} that fails with L{UnauthorizedLogin} when presented with credentials that are unknown to that checker. """ self.options.parseOptions(["--auth", "file:" + self.filename]) checker = self.options["credCheckers"][-1] invalid = UsernamePassword(self.usernamePassword[0], "fake") # Wrong password should raise error return self.assertFailure( checker.requestAvatarId(invalid), error.UnauthorizedLogin ) def test_authSuccess(self): """ The checker created by the C{--auth} command-line option returns a L{Deferred} that returns the avatar id when presented with credentials that are known to that checker. """ self.options.parseOptions(["--auth", "file:" + self.filename]) checker = self.options["credCheckers"][-1] correct = UsernamePassword(*self.usernamePassword) d = checker.requestAvatarId(correct) def checkSuccess(username): self.assertEqual(username, correct.username) return d.addCallback(checkSuccess) def test_checkers(self): """ The L{OpenSSHFactory} built by L{tap.makeService} has a portal with L{ISSHPrivateKey} and L{IUsernamePassword} interfaces registered as checkers. """ config = tap.Options() service = tap.makeService(config) portal = service.factory.portal self.assertEqual( set(portal.checkers.keys()), {ISSHPrivateKey, IUsernamePassword} ) usr/lib/python3/dist-packages/twisted/names/test/test_tap.py 0000644 00000011222 15030104527 0020203 0 ustar 00 # Copyright (c) Twisted Matrix Laboratories. # See LICENSE for details. """ Tests for L{twisted.names.tap}. """ from twisted.internet.base import ThreadedResolver from twisted.names.client import Resolver from twisted.names.dns import PORT from twisted.names.resolve import ResolverChain from twisted.names.secondary import SecondaryAuthorityService from twisted.names.tap import Options, _buildResolvers from twisted.python.runtime import platform from twisted.python.usage import UsageError from twisted.trial.unittest import SynchronousTestCase class OptionsTests(SynchronousTestCase): """ Tests for L{Options}, defining how command line arguments for the DNS server are parsed. """ def test_malformedSecondary(self): """ If the value supplied for an I{--secondary} option does not provide a server IP address, optional port number, and domain name, L{Options.parseOptions} raises L{UsageError}. """ options = Options() self.assertRaises(UsageError, options.parseOptions, ["--secondary", ""]) self.assertRaises(UsageError, options.parseOptions, ["--secondary", "1.2.3.4"]) self.assertRaises( UsageError, options.parseOptions, ["--secondary", "1.2.3.4:hello"] ) self.assertRaises( UsageError, options.parseOptions, ["--secondary", "1.2.3.4:hello/example.com"], ) def test_secondary(self): """ An argument of the form C{"ip/domain"} is parsed by L{Options} for the I{--secondary} option and added to its list of secondaries, using the default DNS port number. """ options = Options() options.parseOptions(["--secondary", "1.2.3.4/example.com"]) self.assertEqual([(("1.2.3.4", PORT), ["example.com"])], options.secondaries) def test_secondaryExplicitPort(self): """ An argument of the form C{"ip:port/domain"} can be used to specify an alternate port number for which to act as a secondary. """ options = Options() options.parseOptions(["--secondary", "1.2.3.4:5353/example.com"]) self.assertEqual([(("1.2.3.4", 5353), ["example.com"])], options.secondaries) def test_secondaryAuthorityServices(self): """ After parsing I{--secondary} options, L{Options} constructs a L{SecondaryAuthorityService} instance for each configured secondary. """ options = Options() options.parseOptions( [ "--secondary", "1.2.3.4:5353/example.com", "--secondary", "1.2.3.5:5354/example.com", ] ) self.assertEqual(len(options.svcs), 2) secondary = options.svcs[0] self.assertIsInstance(options.svcs[0], SecondaryAuthorityService) self.assertEqual(secondary.primary, "1.2.3.4") self.assertEqual(secondary._port, 5353) secondary = options.svcs[1] self.assertIsInstance(options.svcs[1], SecondaryAuthorityService) self.assertEqual(secondary.primary, "1.2.3.5") self.assertEqual(secondary._port, 5354) def test_recursiveConfiguration(self): """ Recursive DNS lookups, if enabled, should be a last-resort option. Any other lookup method (cache, local lookup, etc.) should take precedence over recursive lookups """ options = Options() options.parseOptions(["--hosts-file", "hosts.txt", "--recursive"]) ca, cl = _buildResolvers(options) # Extra cleanup, necessary on POSIX because client.Resolver doesn't know # when to stop parsing resolv.conf. See #NNN for improving this. for x in cl: if isinstance(x, ResolverChain): recurser = x.resolvers[-1] if isinstance(recurser, Resolver): recurser._parseCall.cancel() # On Windows, we need to use a threaded resolver, which leaves trash # lying about that we can't easily clean up without reaching into the # reactor and cancelling them. We only cancel the cleanup functions, as # there should be no others (and it leaving a callLater lying about # should rightly cause the test to fail). if platform.getType() != "posix": # We want the delayed calls on the reactor, which should be all of # ours from the threaded resolver cleanup from twisted.internet import reactor for x in reactor._newTimedCalls: self.assertEqual(x.func.__func__, ThreadedResolver._cleanup) x.cancel() self.assertIsInstance(cl[-1], ResolverChain)
| ver. 1.4 |
Github
|
.
| PHP 8.2.28 | Generation time: 0.02 |
proxy
|
phpinfo
|
Settings