gce_firewall_enforcer_test.py 文件源码

def test_apply_change_lots_of_rules(self):
        """Changing more rules than permitted by the operation semaphore works.

        Setup:
          * Create a new bounded semaphore with a limit of 2 operations.
          * Create a list of 10 rules to insert.
          * Run _apply_change.

        Expected Results:
          * All rules end up in the successes list.
        """
        insert_function = self.firewall_api.insert_firewall_rule
        self.enforcer.operation_sema = threading.BoundedSemaphore(value=2)

        test_rule_name = 'test-network-allow-internal-0'
        test_rule = constants.EXPECTED_FIREWALL_RULES[test_rule_name]

        test_rules = []
        for i in xrange(10):
            rule = copy.deepcopy(test_rule)
            rule['name'] = '%s-%i' % (test_rule_name, i)
            test_rules.append(rule)

        (successes, failures, change_errors) = self.enforcer._apply_change(
            insert_function, test_rules)
        self.assertSameStructure(test_rules, successes)
        self.assertListEqual([], failures)
        self.assertListEqual([], change_errors)