summaryrefslogtreecommitdiff
path: root/source4/scripting/python
diff options
context:
space:
mode:
Diffstat (limited to 'source4/scripting/python')
-rw-r--r--source4/scripting/python/config.mk4
-rw-r--r--source4/scripting/python/samba/provision.py2
-rw-r--r--source4/scripting/python/samba/tests/dcerpc/rpcecho.py3
-rw-r--r--source4/scripting/python/subunit/__init__.py388
4 files changed, 6 insertions, 391 deletions
diff --git a/source4/scripting/python/config.mk b/source4/scripting/python/config.mk
index b494ee6e8d..c88728a9fd 100644
--- a/source4/scripting/python/config.mk
+++ b/source4/scripting/python/config.mk
@@ -23,7 +23,7 @@ python_misc_OBJ_FILES = $(pyscriptsrcdir)/misc_wrap.o
$(python_misc_OBJ_FILES): CFLAGS+=$(CFLAG_NO_UNUSED_MACROS) $(CFLAG_NO_CAST_QUAL)
-_PY_FILES = $(shell find $(pyscriptsrcdir)/samba $(pyscriptsrcdir)/subunit -name "*.py")
+_PY_FILES = $(shell find $(pyscriptsrcdir)/samba ../lib/subunit/python -name "*.py")
$(foreach pyfile, $(_PY_FILES),$(eval $(call python_py_module_template,$(patsubst $(pyscriptsrcdir)/%,%,$(pyfile)),$(pyfile))))
@@ -32,6 +32,6 @@ $(eval $(call python_py_module_template,samba/misc.py,$(pyscriptsrcdir)/misc.py)
EPYDOC_OPTIONS = --no-private --url http://www.samba.org/ --no-sourcecode
epydoc:: pythonmods
- PYTHONPATH=$(pythonbuilddir) epydoc $(EPYDOC_OPTIONS) samba tdb ldb subunit
+ PYTHONPATH=$(pythonbuilddir):../lib/subunit/python epydoc $(EPYDOC_OPTIONS) samba tdb ldb subunit
install:: installpython
diff --git a/source4/scripting/python/samba/provision.py b/source4/scripting/python/samba/provision.py
index 68f61532ad..427ca975b3 100644
--- a/source4/scripting/python/samba/provision.py
+++ b/source4/scripting/python/samba/provision.py
@@ -744,6 +744,8 @@ def setup_samdb(path, setup_path, session_info, credentials, lp,
samdb = SamDB(path, session_info=session_info,
credentials=credentials, lp=lp)
+ if fill == FILL_DRS:
+ return samdb
message("Pre-loading the Samba 4 and AD schema")
samdb.set_domain_sid(domainsid)
diff --git a/source4/scripting/python/samba/tests/dcerpc/rpcecho.py b/source4/scripting/python/samba/tests/dcerpc/rpcecho.py
index 12638e2397..06790736b8 100644
--- a/source4/scripting/python/samba/tests/dcerpc/rpcecho.py
+++ b/source4/scripting/python/samba/tests/dcerpc/rpcecho.py
@@ -25,7 +25,8 @@ from samba.tests import RpcInterfaceTestCase
class RpcEchoTests(RpcInterfaceTestCase):
def setUp(self):
- self.conn = echo.rpcecho("ncalrpc:", self.get_loadparm())
+ self.conn = echo.rpcecho("ncalrpc:", self.get_loadparm(),
+ self.get_credentials())
def test_two_contexts(self):
self.conn2 = echo.rpcecho("ncalrpc:", self.get_loadparm(), basis_connection=self.conn)
diff --git a/source4/scripting/python/subunit/__init__.py b/source4/scripting/python/subunit/__init__.py
deleted file mode 100644
index 406cd8765b..0000000000
--- a/source4/scripting/python/subunit/__init__.py
+++ /dev/null
@@ -1,388 +0,0 @@
-#
-# subunit: extensions to python unittest to get test results from subprocesses.
-# Copyright (C) 2005 Robert Collins <robertc@robertcollins.net>
-# Copyright (C) 2007 Jelmer Vernooij <jelmer@samba.org>
-#
-# This program is free software; you can redistribute it and/or modify
-# it under the terms of the GNU General Public License as published by
-# the Free Software Foundation; either version 3 of the License, or
-# (at your option) any later version.
-#
-# This program is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-# GNU General Public License for more details.
-#
-# You should have received a copy of the GNU General Public License
-# along with this program; if not, write to the Free Software
-# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
-#
-
-import os
-from StringIO import StringIO
-import sys
-import unittest
-
-def test_suite():
- import subunit.tests
- return subunit.tests.test_suite()
-
-
-def join_dir(base_path, path):
- """
- Returns an absolute path to C{path}, calculated relative to the parent
- of C{base_path}.
-
- @param base_path: A path to a file or directory.
- @param path: An absolute path, or a path relative to the containing
- directory of C{base_path}.
-
- @return: An absolute path to C{path}.
- """
- return os.path.join(os.path.dirname(os.path.abspath(base_path)), path)
-
-
-class TestProtocolServer(object):
- """A class for receiving results from a TestProtocol client."""
-
- OUTSIDE_TEST = 0
- TEST_STARTED = 1
- READING_FAILURE = 2
- READING_ERROR = 3
-
- def __init__(self, client, stream=sys.stdout):
- """Create a TestProtocol server instance.
-
- client should be an object that provides
- - startTest
- - addSuccess
- - addFailure
- - addError
- - stopTest
- methods, i.e. a TestResult.
- """
- self.state = TestProtocolServer.OUTSIDE_TEST
- self.client = client
- self._stream = stream
-
- def _addError(self, offset, line):
- if (self.state == TestProtocolServer.TEST_STARTED and
- self.current_test_description == line[offset:-1]):
- self.state = TestProtocolServer.OUTSIDE_TEST
- self.current_test_description = None
- self.client.addError(self._current_test, RemoteError(""))
- self.client.stopTest(self._current_test)
- self._current_test = None
- elif (self.state == TestProtocolServer.TEST_STARTED and
- self.current_test_description + " [" == line[offset:-1]):
- self.state = TestProtocolServer.READING_ERROR
- self._message = ""
- else:
- self.stdOutLineReceived(line)
-
- def _addFailure(self, offset, line):
- if (self.state == TestProtocolServer.TEST_STARTED and
- self.current_test_description == line[offset:-1]):
- self.state = TestProtocolServer.OUTSIDE_TEST
- self.current_test_description = None
- self.client.addFailure(self._current_test, RemoteError())
- self.client.stopTest(self._current_test)
- elif (self.state == TestProtocolServer.TEST_STARTED and
- self.current_test_description + " [" == line[offset:-1]):
- self.state = TestProtocolServer.READING_FAILURE
- self._message = ""
- else:
- self.stdOutLineReceived(line)
-
- def _addSuccess(self, offset, line):
- if (self.state == TestProtocolServer.TEST_STARTED and
- self.current_test_description == line[offset:-1]):
- self.client.addSuccess(self._current_test)
- self.client.stopTest(self._current_test)
- self.current_test_description = None
- self._current_test = None
- self.state = TestProtocolServer.OUTSIDE_TEST
- else:
- self.stdOutLineReceived(line)
-
- def _appendMessage(self, line):
- if line[0:2] == " ]":
- # quoted ] start
- self._message += line[1:]
- else:
- self._message += line
-
- def endQuote(self, line):
- if self.state == TestProtocolServer.READING_FAILURE:
- self.state = TestProtocolServer.OUTSIDE_TEST
- self.current_test_description = None
- self.client.addFailure(self._current_test,
- RemoteError(self._message))
- self.client.stopTest(self._current_test)
- elif self.state == TestProtocolServer.READING_ERROR:
- self.state = TestProtocolServer.OUTSIDE_TEST
- self.current_test_description = None
- self.client.addError(self._current_test,
- RemoteError(self._message))
- self.client.stopTest(self._current_test)
- else:
- self.stdOutLineReceived(line)
-
- def lineReceived(self, line):
- """Call the appropriate local method for the received line."""
- if line == "]\n":
- self.endQuote(line)
- elif (self.state == TestProtocolServer.READING_FAILURE or
- self.state == TestProtocolServer.READING_ERROR):
- self._appendMessage(line)
- else:
- parts = line.split(None, 1)
- if len(parts) == 2:
- cmd, rest = parts
- offset = len(cmd) + 1
- cmd = cmd.strip(':')
- if cmd in ('test', 'testing'):
- self._startTest(offset, line)
- elif cmd == 'error':
- self._addError(offset, line)
- elif cmd == 'failure':
- self._addFailure(offset, line)
- elif cmd in ('success', 'successful'):
- self._addSuccess(offset, line)
- else:
- self.stdOutLineReceived(line)
- else:
- self.stdOutLineReceived(line)
-
- def lostConnection(self):
- """The input connection has finished."""
- if self.state == TestProtocolServer.TEST_STARTED:
- self.client.addError(self._current_test,
- RemoteError("lost connection during test '%s'"
- % self.current_test_description))
- self.client.stopTest(self._current_test)
- elif self.state == TestProtocolServer.READING_ERROR:
- self.client.addError(self._current_test,
- RemoteError("lost connection during "
- "error report of test "
- "'%s'" %
- self.current_test_description))
- self.client.stopTest(self._current_test)
- elif self.state == TestProtocolServer.READING_FAILURE:
- self.client.addError(self._current_test,
- RemoteError("lost connection during "
- "failure report of test "
- "'%s'" %
- self.current_test_description))
- self.client.stopTest(self._current_test)
-
- def readFrom(self, pipe):
- for line in pipe.readlines():
- self.lineReceived(line)
- self.lostConnection()
-
- def _startTest(self, offset, line):
- """Internal call to change state machine. Override startTest()."""
- if self.state == TestProtocolServer.OUTSIDE_TEST:
- self.state = TestProtocolServer.TEST_STARTED
- self._current_test = RemotedTestCase(line[offset:-1])
- self.current_test_description = line[offset:-1]
- self.client.startTest(self._current_test)
- else:
- self.stdOutLineReceived(line)
-
- def stdOutLineReceived(self, line):
- self._stream.write(line)
-
-
-class RemoteException(Exception):
- """An exception that occured remotely to python."""
-
- def __eq__(self, other):
- try:
- return self.args == other.args
- except AttributeError:
- return False
-
-
-class TestProtocolClient(unittest.TestResult):
- """A class that looks like a TestResult and informs a TestProtocolServer."""
-
- def __init__(self, stream):
- super(TestProtocolClient, self).__init__()
- self._stream = stream
-
- def addError(self, test, error):
- """Report an error in test test."""
- self._stream.write("error: %s [\n" % (test.shortDescription() or str(test)))
- for line in self._exc_info_to_string(error, test).splitlines():
- self._stream.write("%s\n" % line)
- self._stream.write("]\n")
- super(TestProtocolClient, self).addError(test, error)
-
- def addFailure(self, test, error):
- """Report a failure in test test."""
- self._stream.write("failure: %s [\n" % (test.shortDescription() or str(test)))
- for line in self._exc_info_to_string(error, test).splitlines():
- self._stream.write("%s\n" % line)
- self._stream.write("]\n")
- super(TestProtocolClient, self).addFailure(test, error)
-
- def addSuccess(self, test):
- """Report a success in a test."""
- self._stream.write("successful: %s\n" % (test.shortDescription() or str(test)))
- super(TestProtocolClient, self).addSuccess(test)
-
- def startTest(self, test):
- """Mark a test as starting its test run."""
- self._stream.write("test: %s\n" % (test.shortDescription() or str(test)))
- super(TestProtocolClient, self).startTest(test)
-
-
-def RemoteError(description=""):
- if description == "":
- description = "\n"
- return (RemoteException, RemoteException(description), None)
-
-
-class RemotedTestCase(unittest.TestCase):
- """A class to represent test cases run in child processes."""
-
- def __eq__ (self, other):
- try:
- return self.__description == other.__description
- except AttributeError:
- return False
-
- def __init__(self, description):
- """Create a psuedo test case with description description."""
- self.__description = description
-
- def error(self, label):
- raise NotImplementedError("%s on RemotedTestCases is not permitted." %
- label)
-
- def setUp(self):
- self.error("setUp")
-
- def tearDown(self):
- self.error("tearDown")
-
- def shortDescription(self):
- return self.__description
-
- def id(self):
- return "%s.%s" % (self._strclass(), self.__description)
-
- def __str__(self):
- return "%s (%s)" % (self.__description, self._strclass())
-
- def __repr__(self):
- return "<%s description='%s'>" % \
- (self._strclass(), self.__description)
-
- def run(self, result=None):
- if result is None: result = self.defaultTestResult()
- result.startTest(self)
- result.addError(self, RemoteError("Cannot run RemotedTestCases.\n"))
- result.stopTest(self)
-
- def _strclass(self):
- cls = self.__class__
- return "%s.%s" % (cls.__module__, cls.__name__)
-
-
-class ExecTestCase(unittest.TestCase):
- """A test case which runs external scripts for test fixtures."""
-
- def __init__(self, methodName='runTest'):
- """Create an instance of the class that will use the named test
- method when executed. Raises a ValueError if the instance does
- not have a method with the specified name.
- """
- unittest.TestCase.__init__(self, methodName)
- testMethod = getattr(self, methodName)
- self.script = join_dir(sys.modules[self.__class__.__module__].__file__,
- testMethod.__doc__)
-
- def countTestCases(self):
- return 1
-
- def run(self, result=None):
- if result is None: result = self.defaultTestResult()
- self._run(result)
-
- def debug(self):
- """Run the test without collecting errors in a TestResult"""
- self._run(unittest.TestResult())
-
- def _run(self, result):
- protocol = TestProtocolServer(result)
- output = os.popen(self.script, mode='r')
- protocol.readFrom(output)
-
-
-class IsolatedTestCase(unittest.TestCase):
- """A TestCase which runs its tests in a forked process."""
-
- def run(self, result=None):
- if result is None: result = self.defaultTestResult()
- run_isolated(unittest.TestCase, self, result)
-
-
-class IsolatedTestSuite(unittest.TestSuite):
- """A TestCase which runs its tests in a forked process."""
-
- def run(self, result=None):
- if result is None: result = unittest.TestResult()
- run_isolated(unittest.TestSuite, self, result)
-
-
-def run_isolated(klass, self, result):
- """Run a test suite or case in a subprocess, using the run method on klass.
- """
- c2pread, c2pwrite = os.pipe()
- # fixme - error -> result
- # now fork
- pid = os.fork()
- if pid == 0:
- # Child
- # Close parent's pipe ends
- os.close(c2pread)
- # Dup fds for child
- os.dup2(c2pwrite, 1)
- # Close pipe fds.
- os.close(c2pwrite)
-
- # at this point, sys.stdin is redirected, now we want
- # to filter it to escape ]'s.
- ### XXX: test and write that bit.
-
- result = TestProtocolClient(sys.stdout)
- klass.run(self, result)
- sys.stdout.flush()
- sys.stderr.flush()
- # exit HARD, exit NOW.
- os._exit(0)
- else:
- # Parent
- # Close child pipe ends
- os.close(c2pwrite)
- # hookup a protocol engine
- protocol = TestProtocolServer(result)
- protocol.readFrom(os.fdopen(c2pread, 'rU'))
- os.waitpid(pid, 0)
- # TODO return code evaluation.
- return result
-
-
-class SubunitTestRunner(object):
- def __init__(self, stream=sys.stdout):
- self.stream = stream
-
- def run(self, test):
- "Run the given test case or test suite."
- result = TestProtocolClient(self.stream)
- test(result)
- return result
-
generated by cgit (git 2.34.1) at 2024-11-10 08:13:35 +0000