This file is indexed.

/usr/share/pyshared/proboscis/case.py is in python-proboscis 1.2.6.0-1.

This file is owned by root:root, with mode 0o644.

The actual contents of the file can be viewed below.

  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
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
# Copyright (c) 2011 Rackspace
# All Rights Reserved.
#
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.

"""Creates TestCases from a list of TestEntries.

This module mainly exists to translate Proboscis classes and concepts into the
unittest equivalents.

"""

import os
import pydoc
import types
import unittest
import sys

from collections import deque
from functools import wraps

from proboscis import compatability
from proboscis import dependencies
from proboscis import SkipTest
from proboscis.sorting import TestGraph
from proboscis.core import TestMethodClassEntry
from proboscis.decorators import DEFAULT_REGISTRY

# This is here so Proboscis own test harness can change it while still calling
# TestProgram normally. Its how the examples are tested.
OVERRIDE_DEFAULT_STREAM = None


class TestPlan(object):
    """Grabs information from the TestRegistry and creates a test plan."""

    def __init__(self, groups, test_entries, factories):
        test_cases = self.create_cases(test_entries, factories)
        graph = TestGraph(groups, test_entries, test_cases)
        self.tests = graph.sort()

    @staticmethod
    def create_from_registry(registry):
        """Returns a sorted TestPlan from a TestRegistry instance."""
        return TestPlan(registry.groups, registry.tests, registry.factories)

    @staticmethod
    def create_cases_from_instance(factory, instance):
        if isinstance(instance, type):
            raise RuntimeError("Factory %s returned type %s (rather than an "
                "instance), which is not allowed." % (factory, instance))
        if isinstance(instance, types.MethodType):
            home = compatability.get_method_function(instance)
        elif isinstance(instance, types.FunctionType):
            home = instance
        else:
            home = type(instance)
        if issubclass(home, unittest.TestCase):
            raise RuntimeError("Factory %s returned a unittest.TestCase "
                "instance %s, which is not legal.")
        try:
            entry = home._proboscis_entry_
        except AttributeError:
            raise RuntimeError("Factory method %s returned an instance %s "
                "which was not tagged as a Proboscis TestEntry." %
                (factory, instance))
        entry.mark_as_used_by_factory()  # Don't iterate this since a
                                         # function is creating it.
        if entry.is_child:
            raise RuntimeError("Function %s, which exists as a bound method "
                "in a decorated class may not be returned from a factory." %
                instance)
        # There is potentially an issue in that a different Registry might
        # register an entry, and we could then read that in with a factory.
        # Later the entry would not be found in the dictionary of entries.
        if isinstance(instance, types.MethodType):
            try:
                state = TestMethodState(instance.im_self)
            except AttributeError:
                raise RuntimeError("Only bound methods may be returned from "
                    "factories. %s is not bound." % instance)
        else:
            state = TestMethodState(entry, instance)
        return TestPlan._create_test_cases_for_entry(entry, state)

    @staticmethod
    def create_cases(test_entries, factories):
        tests = []
        entries = {}
        for factory in factories:
            list = factory()
            for item in list:
                cases = TestPlan.create_cases_from_instance(factory, item)
                tests += cases
        for entry in test_entries:
            if not entry.is_child and not entry.used_by_factory:
                test_cases = TestPlan._create_test_cases_for_entry(entry)
                entries[entry] = test_cases
                tests += test_cases
        return tests

    @staticmethod
    def _create_test_cases_for_entry(entry, state=None):
        """Processes a test case entry."""
        if not hasattr(entry, 'children'):  # function or unittest.TestCase
            return [TestCase(entry)]
        state = state or TestMethodState(entry)
        cases = []
        for child_entry in entry.children:
            case = TestCase(child_entry, state=state)
            cases.append(case)
        return cases

    def create_test_suite(self, config, loader):
        """Transforms the plan into a Nose test suite."""
        creator = TestSuiteCreator(loader)
        if dependencies.use_nose:
            from nose.suite import ContextSuiteFactory
            suite = ContextSuiteFactory(config)([])
        else:
            suite = unittest.TestSuite()
        for case in self.tests:
            if case.entry.info.enabled and case.entry.home is not None:
                tests = creator.loadTestsFromTestEntry(case)
                for test in tests:
                    suite.addTest(test)
        return suite

    def filter(self, group_names=None, classes=None, functions=None):
        """Whittles down test list to those matching criteria."""
        test_homes = []
        classes = classes or []
        functions = functions or []
        for cls in classes:
            test_homes.append(cls)
        for function in functions:
            test_homes.append(function)
        group_names = group_names or []
        filtered_list = []
        while self.tests:
            case = self.tests.pop()
            if case.entry.contains(group_names, test_homes):
                filtered_list.append(case)
                # Add any groups this depends on so they will run as well.
                for group_name in case.entry.info.depends_on_groups:
                    if not group_name in group_names:
                        group_names.append(group_name)
                for test_home in case.entry.info.depends_on:
                    if not test_home in test_homes:
                        test_homes.append(test_home)
        self.tests = list(reversed(filtered_list))


class TestCase(object):
    """Represents an instance of a TestEntry.

    This class is also used to store status information, such as the dependent
    TestEntry objects (discovered when this test is sorted) and any failure
    in the dependencies of this test (used to raise SkipTest if needed).

    There may be multiple TestCase instances for each TestEntry instance.

    """
    def __init__(self, entry, state=None):
        self.entry = entry
        self.dependents = []  # This is populated when we sort the tests.
        self.dependency_failure = None
        self.state = state

    def check_dependencies(self, test_self):
        """If a dependency has failed, SkipTest is raised."""
        if self.dependency_failure is not None and \
           self.dependency_failure != self and not self.entry.info.always_run:
            home = self.dependency_failure.entry.home
            dependencies.skip_test(test_self, "Failure in %s" % home)

    def fail_test(self, dependency_failure=None):
        """Called when this entry fails to notify dependents."""
        if not dependency_failure:
            dependency_failure = self
        if not self.dependency_failure:  # Do NOT overwrite the first cause
            self.dependency_failure = dependency_failure
            for dependent in self.dependents:
                if dependent.critical:
                    dependent.case.fail_test(
                        dependency_failure=dependency_failure)

    def write_doc(self, file):
        file.write(str(self.entry.home) + "\n")
        doc = pydoc.getdoc(self.entry.home)
        if doc:
            file.write(doc + "\n")
        for field in str(self.entry.info).split(', '):
            file.write("\t" + field + "\n")

    def __repr__(self):
        return "TestCase(" + repr(self.entry.home) + ", " + \
               repr(self.entry.info) + ", " + object.__repr__(self) + ")"

    def __str__(self):
        return "Home = " + str(self.entry.home) + ", Info(" + \
               str(self.entry.info) + ")"


class TestResultListener():
    """Implements methods of TestResult to be informed of test failures."""

    def __init__(self, chain_to_cls):
        self.chain_to_cls = chain_to_cls

    def addError(self, test, err):
        self.onError(test)
        self.chain_to_cls.addError(self, test, err)

    def addFailure(self, test, err):
        self.onError(test)
        self.chain_to_cls.addFailure(self, test, err)

    def onError(self, test):
        """Notify a test entry and its dependents of failure."""
        if dependencies.use_nose:
            root = test.test
        else:
            root = test
        if hasattr(root, "__proboscis_case__"):
            case = root.__proboscis_case__
            case.fail_test()



class TestResult(TestResultListener, dependencies.TextTestResult):
    """Adds Proboscis skip on dependency failure functionality.

    Extends either Nose or unittest's TextTestResult class.
    If a program needs to use its own TestResult class it must inherit from
    this class and call "onError" at the start of both the addError and
    addFailure functions, passing the "test" parameter, to keep
    Proboscis's skip on depdendency failure functionality.

    """

    # I had issues extending TextTestResult directly so resorted to this.

    def __init__(self, *args, **kwargs):
        TestResultListener.__init__(self, dependencies.TextTestResult)
        dependencies.TextTestResult.__init__(self, *args, **kwargs)


def test_runner_cls(wrapped_cls, cls_name):
    """Creates a test runner class which uses Proboscis TestResult."""
    new_dict = wrapped_cls.__dict__.copy()

    if dependencies.use_nose:
        def cb_make_result(self):
            return TestResult(self.stream, self.descriptions, self.verbosity,
                              self.config)
    else:
        def cb_make_result(self):
            return TestResult(self.stream, self.descriptions, self.verbosity)
    new_dict["_makeResult"] = cb_make_result
    return type(cls_name, (wrapped_cls,), new_dict)


def skippable_func(test_case, func):
    """Gives free functions a Nose independent way of skipping a test.

    The unittest module TestCase class has a skipTest method, but to run it you
    need access to the TestCase class. This wraps the runTest method of the
    underlying unittest.TestCase subclass to invoke the skipTest method if it
    catches the SkipTest exception.

    """
    s_func = None
    if dependencies.use_nose:
        s_func = func
    else:
        @wraps(func)
        def skip_capture_func():
            st = compatability.capture_exception(func, SkipTest)
            if st is not None:
                dependencies.skip_test(test_case, st.message)
        s_func = skip_capture_func

    @wraps(s_func)
    def testng_method_mistake_capture_func():
        compatability.capture_type_error(s_func)

    return testng_method_mistake_capture_func


class FunctionTest(unittest.FunctionTestCase):
    """Wraps a single function as a test runnable by unittest / nose."""

    def __init__(self, test_case):
        func = test_case.entry.home
        _old_setup = None
        if hasattr(func, 'setup'):  # Don't destroy nose-style setup
            _old_setup = func.setup
        def cb_check(cb_self=None):
            test_case.check_dependencies(self)
            if _old_setup is not None:
                _old_setup()
        self.__proboscis_case__ = test_case
        sfunc = skippable_func(self, func)
        unittest.FunctionTestCase.__init__(self, testFunc=sfunc, setUp=cb_check)


class TestMethodState(object):
    """Manages a test class instance used by one or more test methods."""

    def __init__(self, entry, instance=None):
        self.entry = entry
        # This would be a simple "isinstance" but due to the reloading mania
        # needed for Proboscis's documentation tests it has to be a bit
        # weirder.
        if not str(type(self.entry)) == str(TestMethodClassEntry):
            raise RuntimeError("%s is not a TestMethodClassEntry but is a %s."
                               % (self.entry, type(self.entry)))
        self.instance = instance

    def get_state(self):
        if not self.instance:
            self.instance = self.entry.home()
        return self.instance


class MethodTest(unittest.FunctionTestCase):
    """Wraps a method as a test runnable by unittest."""

    def __init__(self, test_case):
        assert test_case.state is not None
        #TODO: Figure out how to attach calls to BeforeMethod and BeforeClass,
        #      AfterMethod and AfterClass. It should be easy enough to
        #      just find them using the TestEntry parent off test_case Entrty.
        def cb_check(cb_self=None):
            test_case.check_dependencies(self)
        @wraps(test_case.entry.home)
        def func(self=None):  # Called by FunctionTestCase
            func = test_case.entry.home
            func(test_case.state.get_state())
        self.__proboscis_case__ = test_case
        sfunc = skippable_func(self, func)
        unittest.FunctionTestCase.__init__(self, testFunc=sfunc, setUp=cb_check)


def decorate_class(setUp_method=None, tearDown_method=None):
    """Inserts method calls in the setUp / tearDown methods of a class."""
    def return_method(cls):
        """Returns decorated class."""
        new_dict = cls.__dict__.copy()
        if setUp_method:
            if hasattr(cls, "setUp"):
                @wraps(setUp_method)
                def _setUp(self):
                    setUp_method(self)
                    cls.setUp(self)
            else:
                @wraps(setUp_method)
                def _setUp(self):
                    setUp_method(self)
            new_dict["setUp"] = _setUp
        if tearDown_method:
            if hasattr(cls, "tearDown"):
                @wraps(tearDown_method)
                def _tearDown(self):
                    tearDown_method(self)
                    cls.setUp(self)
            else:
                @wraps(tearDown_method)
                def _tearDown(self):
                    tearDown_method(self)
            new_dict["tearDown"] = _tearDown
        return type(cls.__name__, (cls,), new_dict)
    return return_method


class TestSuiteCreator(object):
    """Turns Proboscis test cases into elements to be run by unittest."""

    def __init__(self, loader):
        self.loader = loader

    def loadTestsFromTestEntry(self, test_case):
        """Wraps a test class in magic so it will skip on dependency failures.

        Decorates the testEntry class's setUp method to raise SkipTest if
        tests this test was dependent on failed or had errors.

        """
        home = test_case.entry.home
        if home is None:
            return []
        if isinstance(home, type):
            return self.wrap_unittest_test_case_class(test_case)
        if isinstance(home, types.FunctionType):
            if home._proboscis_entry_.is_child:
                return self.wrap_method(test_case)
            else:
                return self.wrap_function(test_case)
        raise RuntimeError("Unknown test type:" + str(type(home)))

    def wrap_function(self, test_case):
        return [FunctionTest(test_case)]

    def wrap_method(self, test_case):
        return [MethodTest(test_case)]

    def wrap_unittest_test_case_class(self, test_case):
        original_cls = test_case.entry.home
        def cb_check(cb_self):
            test_case.check_dependencies(cb_self)
        testCaseClass = decorate_class(setUp_method=cb_check)(original_cls)
        testCaseNames = self.loader.getTestCaseNames(testCaseClass)
        if not testCaseNames and hasattr(testCaseClass, 'runTest'):
            testCaseNames = ['runTest']
        suite = []
        if issubclass(original_cls, unittest.TestCase):
            for name in testCaseNames:
                test_instance = testCaseClass(name)
                setattr(test_instance, "__proboscis_case__", test_case)
                suite.append(test_instance)
        return suite


class TestProgram(dependencies.TestProgram):
    """Use this to run Proboscis.

    Translates the Proboscis test registry into types used by Nose or unittest
    in order to run the program.

    Most arguments to this are simply passed to Nose or unittest's TestProgram
    class.

    For most cases using the default arguments works fine.

    :param registry: The test registry to use. If unset uses the default global
                     registry.
    :param groups: A list of strings representing the groups of tests to run.
                   The list is added to by parsing the argv argument. If unset
                   then all groups are run.
    :param testLoader: The test loader. By default, its unittest.TestLoader.
    :param config: The config passed to Nose or unittest.TestProgram. The
                   config determines things such as plugins or output streams,
                   so it may be necessary to create this for advanced use
                   cases.
    :param plugins: Nose plugins. Similar to config it may be necessary to
                    set this in an advanced setup.
    :param env: By default is os.environ. This is used only by Nose.
    :param testRunner: By default Proboscis uses its own. If this is set
                       however care must be taken to avoid breaking Proboscis's
                       automatic skipping of tests on dependency failures.
                       In particular, _makeResult must return a subclass of
                       proboscis.TestResult which calls
                       proboscis.TestResult.onError at the start of the
                       addFailure and addError methods.
    :param stream: By default this is standard out.
    :param argv: By default this is sys.argv. Proboscis parses this for the
                 --group argument.
    """
    def __init__(self,
                 registry=DEFAULT_REGISTRY,
                 groups=None,
                 testLoader=None,
                 config=None,
                 plugins=None,
                 env=None,
                 testRunner=None,
                 stream=None,
                 argv=None,
                 *args, **kwargs):
        groups = groups or []
        argv = argv or sys.argv
        argv = self.extract_groups_from_argv(argv, groups)
        if "suite" in kwargs:
            raise ValueError("'suite' is not a valid argument, as Proboscis " \
                             "creates the suite.")

        self.__loader = testLoader or unittest.TestLoader()

        if OVERRIDE_DEFAULT_STREAM:
            stream = OVERRIDE_DEFAULT_STREAM

        if env is None:
            env = os.environ
        if dependencies.use_nose and config is None:
            config = self.makeConfig(env, plugins)
            if not stream:
                stream = config.stream

        stream = stream or sys.stdout

        if testRunner is None:
            runner_cls = test_runner_cls(dependencies.TextTestRunner,
                                         "ProboscisTestRunner")
            if dependencies.use_nose:
                testRunner = runner_cls(stream,
                                        verbosity=3,  # config.verbosity,
                                        config=config)
            else:
                testRunner = runner_cls(stream, verbosity=3)

        #registry.sort()
        self.plan = TestPlan.create_from_registry(registry)

        if len(groups) > 0:
            self.plan.filter(group_names=groups)
        self.cases = self.plan.tests
        if "--show-plan" in argv:
            self.__run = self.show_plan
        else:
            self.__suite = self.create_test_suite_from_entries(config,
                                                               self.cases)
            def run():
                if dependencies.use_nose:
                    dependencies.TestProgram.__init__(
                        self,
                        suite=self.__suite,
                        config=config,
                        env=env,
                        plugins=plugins,
                        testLoader=testLoader,  # Pass arg, not what we create
                        testRunner=testRunner,
                        argv=argv,
                        *args, **kwargs
                    )
                else:
                    dependencies.TestProgram.__init__(
                        self,
                        suite=self.__suite,
                        config=config,
                        testLoader=testLoader,  # Pass arg, not what we create
                        testRunner=testRunner,
                        argv=argv,
                        *args, **kwargs
                    )
            self.__run = run

    def create_test_suite_from_entries(self, config, cases):
        """Creates a suite runnable by unittest."""
        return self.plan.create_test_suite(config, self.__loader)

    def extract_groups_from_argv(self, argv, groups):
        """Given argv, records the "--group" options.

        :param argv: A list of arguments, such as sys.argv.
        :param groups: A list of strings for each group to run which is added
                       to.

        Returns a copy of param argv with the --group options removed. This is
        useful if argv needs to be passed to another program such as Nose.

        """
        new_argv = [argv[0]]
        for arg in argv[1:]:
            if arg[:8] == "--group=":
                groups.append(arg[8:])
            else:
                new_argv.append(arg)
        return new_argv

    def run_and_exit(self):
        """Calls unittest or Nose to run all tests.

        unittest will call sys.exit on completion.

        """
        self.__run()

    def show_plan(self):
        """Prints information on test entries and the order they will run."""
        print("   *  *  *  Test Plan  *  *  *")
        for case in self.cases:
            case.write_doc(sys.stdout)

    @property
    def test_suite(self):
        return self.__suite