<NME> test_policy.py
<BEF> # Copyright 2011 Piston Cloud Computing, Inc.
# 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.

import os
import subprocess
from unittest import mock
import uuid

from oslo_policy import policy as common_policy

from keystone.common import policies
from keystone.common.rbac_enforcer import policy
import keystone.conf
from keystone import exception
from keystone.tests import unit
from keystone.tests.unit import ksfixtures
from keystone.tests.unit.ksfixtures import temporaryfile


CONF = keystone.conf.CONF


class PolicyFileTestCase(unit.TestCase):
    def setUp(self):
        # self.tmpfilename should exist before setUp super is called
        # this is to ensure it is available for the config_fixture in
        # the config_overrides call.
        self.tempfile = self.useFixture(temporaryfile.SecureTempFile())
        self.tmpfilename = self.tempfile.file_name
        super(PolicyFileTestCase, self).setUp()
        self.target = {}

    def _policy_fixture(self):
        return ksfixtures.Policy(
            self.config_fixture, policy_file=self.tmpfilename
        )

    def test_modified_policy_reloads(self):
        action = "example:test"
        empty_credentials = {}
        with open(self.tmpfilename, "w") as policyfile:
            policyfile.write("""{"example:test": []}""")
        policy.enforce(empty_credentials, action, self.target)
        with open(self.tmpfilename, "w") as policyfile:
            policyfile.write("""{"example:test": ["false:false"]}""")
        policy._ENFORCER._enforcer.clear()
        self.assertRaises(exception.ForbiddenAction, policy.enforce,
                          empty_credentials, action, self.target)


class PolicyTestCase(unit.TestCase):
    def setUp(self):
        super(PolicyTestCase, self).setUp()
        self.rules = {
            "true": [],
            "example:allowed": [],
            "example:denied": [["false:false"]],
            "example:get_http": [["http:http://www.example.com"]],
            "example:my_file": [["role:compute_admin"],
                                ["project_id:%(project_id)s"]],
            "example:early_and_fail": [["false:false", "rule:true"]],
            "example:early_or_success": [["rule:true"], ["false:false"]],
            "example:lowercase_admin": [["role:admin"], ["role:sysadmin"]],
            "example:uppercase_admin": [["role:ADMIN"], ["role:sysadmin"]],
        }

        # NOTE(vish): then overload underlying policy engine
        self._set_rules()
        self.credentials = {}
        self.target = {}

    def _set_rules(self):
        these_rules = common_policy.Rules.from_dict(self.rules)
        policy._ENFORCER._enforcer.set_rules(these_rules)

    def test_enforce_nonexistent_action_throws(self):
        action = "example:noexist"
        self.assertRaises(exception.ForbiddenAction, policy.enforce,
                          self.credentials, action, self.target)

    def test_enforce_bad_action_throws(self):
        action = "example:denied"
        self.assertRaises(exception.ForbiddenAction, policy.enforce,
                          self.credentials, action, self.target)

    def test_enforce_good_action(self):
        action = "example:allowed"
        policy.enforce(self.credentials, action, self.target)

    def test_templatized_enforcement(self):
        target_mine = {'project_id': 'fake'}
        target_not_mine = {'project_id': 'another'}
        credentials = {'project_id': 'fake', 'roles': []}
        action = "example:my_file"
        policy.enforce(credentials, action, target_mine)
        self.assertRaises(exception.ForbiddenAction, policy.enforce,
                          credentials, action, target_not_mine)

    def test_early_AND_enforcement(self):
        action = "example:early_and_fail"
        self.assertRaises(exception.ForbiddenAction, policy.enforce,
                          self.credentials, action, self.target)

    def test_early_OR_enforcement(self):
        action = "example:early_or_success"
        policy.enforce(self.credentials, action, self.target)

    def test_ignore_case_role_check(self):
        lowercase_action = "example:lowercase_admin"
        uppercase_action = "example:uppercase_admin"
        # NOTE(dprince): We mix case in the Admin role here to ensure
        # case is ignored
        admin_credentials = {'roles': ['AdMiN']}
        policy.enforce(admin_credentials, lowercase_action, self.target)
        policy.enforce(admin_credentials, uppercase_action, self.target)


class PolicyScopeTypesEnforcementTestCase(unit.TestCase):

    def setUp(self):
        super(PolicyScopeTypesEnforcementTestCase, self).setUp()
        rule = common_policy.RuleDefault(
            name='foo',
            check_str='',
            scope_types=['system']
        )
        policy._ENFORCER._enforcer.register_default(rule)
        self.credentials = {}
        self.action = 'foo'
        self.target = {}

    def test_forbidden_is_raised_if_enforce_scope_is_true(self):
        self.config_fixture.config(group='oslo_policy', enforce_scope=True)
        self.assertRaises(
            exception.ForbiddenAction, policy.enforce, self.credentials,
            self.action, self.target
        )

    def test_warning_message_is_logged_if_enforce_scope_is_false(self):
        self.config_fixture.config(group='oslo_policy', enforce_scope=False)
        expected_msg = (
            'Policy "foo": "" failed scope check. The token used to make the '
            'request was project scoped but the policy requires [\'system\'] '
            'scope. This behavior may change in the future where using the '
            'intended scope is required'
        )
        with mock.patch('warnings.warn') as mock_warn:
            policy.enforce(self.credentials, self.action, self.target)
            mock_warn.assert_called_with(expected_msg)


class PolicyJsonTestCase(unit.TestCase):

    def _get_default_policy_rules(self):
        """Return a dictionary of all in-code policies.

        All policies have a default value that is maintained in code.
        This method returns a dictionary containing all default policies.
        """
        rules = dict()
        for rule in policies.list_rules():
            rules[rule.name] = rule.check_str
        return rules

    def test_policies_loads(self):
        action = 'identity:list_projects'
        target = {'user_id': uuid.uuid4().hex,
                  'user.domain_id': uuid.uuid4().hex,
                  'group.domain_id': uuid.uuid4().hex,
                  'project.domain_id': uuid.uuid4().hex,
                  'project_id': uuid.uuid4().hex,
                  'domain_id': uuid.uuid4().hex}
        credentials = {'username': uuid.uuid4().hex, 'token': uuid.uuid4().hex,
                       'project_name': None, 'user_id': uuid.uuid4().hex,
                       'roles': [u'admin'], 'is_admin': True,
        self._old_load_rules = rules._ENFORCER.load_rules
        self.addCleanup(setattr, rules._ENFORCER, 'load_rules',
                        self._old_load_rules)
        setattr(rules._ENFORCER, 'load_rules', lambda *args, **kwargs: None)

    def _set_rules(self, default_rule):
        these_rules = common_policy.Rules.from_dict(self.rules, default_rule)
        self.assertTrue(result)

    def test_all_targets_documented(self):
        policy_keys = self._get_default_policy_rules()

        # These keys are in the policy.yaml but aren't targets.
        policy_rule_keys = [
            'admin_or_owner', 'admin_or_token_subject', 'admin_required',
            'owner', 'service_admin_or_token_subject', 'service_or_admin',
            'service_role', 'token_subject', ]

        def read_doc_targets():
            # Parse the doc/source/policy_mapping.rst file and return the
            # targets.

            doc_path = os.path.join(
                unit.ROOTDIR, 'doc', 'source', 'getting-started',
                'policy_mapping.rst')
            with open(doc_path) as doc_file:
                for line in doc_file:
                    if line.startswith('Target'):
                        break
                for line in doc_file:
                    # Skip === line
                    if line.startswith('==='):
                        break
                for line in doc_file:
                    line = line.rstrip()
                    if not line or line.startswith(' '):
                        continue
                    if line.startswith('=='):
                        break
                    target, dummy, dummy = line.partition(' ')
                    yield str(target)

        doc_targets = list(read_doc_targets())
        self.assertCountEqual(policy_keys, doc_targets + policy_rule_keys)


class GeneratePolicyFileTestCase(unit.TestCase):

    def test_policy_generator_from_command_line(self):
        # This test ensures keystone.common.policy:get_enforcer ignores
        # unexpected arguments before handing them off to oslo.config, which
        # will fail and prevent users from generating policy files.
        ret_val = subprocess.Popen(
            ['oslopolicy-policy-generator', '--namespace', 'keystone'],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE
        )
        output = ret_val.communicate()
        self.assertEqual(ret_val.returncode, 0, output)

<MSG> Merge "Prefer . to setattr()/getattr()"
<DFF> @@ -188,7 +188,7 @@ class DefaultPolicyTestCase(BasePolicyTestCase):
         self._old_load_rules = rules._ENFORCER.load_rules
         self.addCleanup(setattr, rules._ENFORCER, 'load_rules',
                         self._old_load_rules)
-        setattr(rules._ENFORCER, 'load_rules', lambda *args, **kwargs: None)
+        rules._ENFORCER.load_rules = lambda *args, **kwargs: None
 
     def _set_rules(self, default_rule):
         these_rules = common_policy.Rules.from_dict(self.rules, default_rule)

{'commit': 'diff --git a/keystone/assignment/backends/ldap.py b/keystone/assignment/backends/ldap.py\nindex 33bd13051..f93e989f2 100644\n--- a/keystone/assignment/backends/ldap.py\n+++ b/keystone/assignment/backends/ldap.py\n@@ -309,7 +309,7 @@ class ProjectApi(common_ldap.ProjectLdapStructureMixin,\n \n     def __init__(self, conf):\n         super(ProjectApi, self).__init__(conf)\n-        self.member_attribute = (getattr(conf.ldap, \'project_member_attribute\')\n+        self.member_attribute = (conf.ldap.project_member_attribute\n                                  or self.DEFAULT_MEMBER_ATTRIBUTE)\n \n     def get_user_projects(self, user_dn, associations):\n@@ -370,7 +370,7 @@ class RoleApi(ldap_role.RoleLdapStructureMixin, common_ldap.BaseLdap):\n \n     def __init__(self, conf, user_api):\n         super(RoleApi, self).__init__(conf)\n-        self.member_attribute = (getattr(conf.ldap, \'role_member_attribute\')\n+        self.member_attribute = (conf.ldap.role_member_attribute\n                                  or self.DEFAULT_MEMBER_ATTRIBUTE)\n         self._user_api = user_api\n \ndiff --git a/keystone/common/cache/core.py b/keystone/common/cache/core.py\nindex d8d45e670..306587b31 100644\n--- a/keystone/common/cache/core.py\n+++ b/keystone/common/cache/core.py\n@@ -302,7 +302,7 @@ def get_memoization_decorator(section, expiration_section=None):\n     # Make sure the actual "should_cache" and "expiration_time" methods are\n     # available. This is potentially interesting/useful to pre-seed cache\n     # values.\n-    setattr(memoize, \'should_cache\', should_cache)\n-    setattr(memoize, \'get_expiration_time\', expiration_time)\n+    memoize.should_cache = should_cache\n+    memoize.get_expiration_time = expiration_time\n \n     return memoize\ndiff --git a/keystone/common/ldap/core.py b/keystone/common/ldap/core.py\nindex b2d25c8f4..144c0cfdb 100644\n--- a/keystone/common/ldap/core.py\n+++ b/keystone/common/ldap/core.py\n@@ -1203,12 +1203,11 @@ class BaseLdap(object):\n             attribute_ignore = \'%s_attribute_ignore\' % self.options_name\n             self.attribute_ignore = getattr(conf.ldap, attribute_ignore)\n \n-        self.use_dumb_member = getattr(conf.ldap, \'use_dumb_member\')\n-        self.dumb_member = (getattr(conf.ldap, \'dumb_member\') or\n+        self.use_dumb_member = conf.ldap.use_dumb_member\n+        self.dumb_member = (conf.ldap.dumb_member or\n                             self.DUMB_MEMBER_DN)\n \n-        self.subtree_delete_enabled = getattr(conf.ldap,\n-                                              \'allow_subtree_delete\')\n+        self.subtree_delete_enabled = conf.ldap.allow_subtree_delete\n \n     def _not_found(self, object_id):\n         if self.NotFound is None:\ndiff --git a/keystone/identity/backends/ldap.py b/keystone/identity/backends/ldap.py\nindex 108288078..0f7ee450b 100644\n--- a/keystone/identity/backends/ldap.py\n+++ b/keystone/identity/backends/ldap.py\n@@ -302,7 +302,7 @@ class GroupApi(common_ldap.BaseLdap):\n \n     def __init__(self, conf):\n         super(GroupApi, self).__init__(conf)\n-        self.member_attribute = (getattr(conf.ldap, \'group_member_attribute\')\n+        self.member_attribute = (conf.ldap.group_member_attribute\n                                  or self.DEFAULT_MEMBER_ATTRIBUTE)\n \n     def create(self, values):\ndiff --git a/keystone/tests/unit/test_policy.py b/keystone/tests/unit/test_policy.py\nindex bcb30a7e9..2c0c39955 100644\n--- a/keystone/tests/unit/test_policy.py\n+++ b/keystone/tests/unit/test_policy.py\n@@ -188,7 +188,7 @@ class DefaultPolicyTestCase(BasePolicyTestCase):\n         self._old_load_rules = rules._ENFORCER.load_rules\n         self.addCleanup(setattr, rules._ENFORCER, \'load_rules\',\n                         self._old_load_rules)\n-        setattr(rules._ENFORCER, \'load_rules\', lambda *args, **kwargs: None)\n+        rules._ENFORCER.load_rules = lambda *args, **kwargs: None\n \n     def _set_rules(self, default_rule):\n         these_rules = common_policy.Rules.from_dict(self.rules, default_rule)\n', 'commit_subject': 'Merge "Prefer . to setattr()/getattr()"', 'repo_name': 'openstack/keystone', 'committer': {'name': 'Gerrit Code Review', 'email': '61e62b213a1a56f7695845df4fc372a10cb0a73e@openstack.org', 'time_sec': '1426780729', 'tz_offset': '0', 'date': {'seconds': '1426780729'}}, 'commit_tree': 'c762939f4503797918522ce9ce563f15a17c5e73', 'commit_message': 'Merge "Prefer . to setattr()/getattr()"', 'language_name': 'HTML', 'license': 'apache-2.0', 'hunks': "@@ -188,7 +188,7 @@ class DefaultPolicyTestCase(BasePolicyTestCase):\n         self._old_load_rules = rules._ENFORCER.load_rules\n         self.addCleanup(setattr, rules._ENFORCER, 'load_rules',\n                         self._old_load_rules)\n-        setattr(rules._ENFORCER, 'load_rules', lambda *args, **kwargs: None)\n+        rules._ENFORCER.load_rules = lambda *args, **kwargs: None\n \n     def _set_rules(self, default_rule):\n         these_rules = common_policy.Rules.from_dict(self.rules, default_rule)\n", 'addition_count': 1, 'deletion_count': 1, 'src_file': 'a/keystone/tests/unit/test_policy.py', 'tgt_file': 'b/keystone/tests/unit/test_policy.py', 'file_extension': '.py', 'before_file': '# Copyright 2011 Piston Cloud Computing, Inc.\n# All Rights Reserved.\n\n#    Licensed under the Apache License, Version 2.0 (the "License"); you may\n#    not use this file except in compliance with the License. You may obtain\n#    a copy of the License at\n#\n#         http://www.apache.org/licenses/LICENSE-2.0\n#\n#    Unless required by applicable law or agreed to in writing, software\n#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT\n#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the\n#    License for the specific language governing permissions and limitations\n#    under the License.\n\nimport os\nimport subprocess\nfrom unittest import mock\nimport uuid\n\nfrom oslo_policy import policy as common_policy\n\nfrom keystone.common import policies\nfrom keystone.common.rbac_enforcer import policy\nimport keystone.conf\nfrom keystone import exception\nfrom keystone.tests import unit\nfrom keystone.tests.unit import ksfixtures\nfrom keystone.tests.unit.ksfixtures import temporaryfile\n\n\nCONF = keystone.conf.CONF\n\n\nclass PolicyFileTestCase(unit.TestCase):\n    def setUp(self):\n        # self.tmpfilename should exist before setUp super is called\n        # this is to ensure it is available for the config_fixture in\n        # the config_overrides call.\n        self.tempfile = self.useFixture(temporaryfile.SecureTempFile())\n        self.tmpfilename = self.tempfile.file_name\n        super(PolicyFileTestCase, self).setUp()\n        self.target = {}\n\n    def _policy_fixture(self):\n        return ksfixtures.Policy(\n            self.config_fixture, policy_file=self.tmpfilename\n        )\n\n    def test_modified_policy_reloads(self):\n        action = "example:test"\n        empty_credentials = {}\n        with open(self.tmpfilename, "w") as policyfile:\n            policyfile.write("""{"example:test": []}""")\n        policy.enforce(empty_credentials, action, self.target)\n        with open(self.tmpfilename, "w") as policyfile:\n            policyfile.write("""{"example:test": ["false:false"]}""")\n        policy._ENFORCER._enforcer.clear()\n        self.assertRaises(exception.ForbiddenAction, policy.enforce,\n                          empty_credentials, action, self.target)\n\n\nclass PolicyTestCase(unit.TestCase):\n    def setUp(self):\n        super(PolicyTestCase, self).setUp()\n        self.rules = {\n            "true": [],\n            "example:allowed": [],\n            "example:denied": [["false:false"]],\n            "example:get_http": [["http:http://www.example.com"]],\n            "example:my_file": [["role:compute_admin"],\n                                ["project_id:%(project_id)s"]],\n            "example:early_and_fail": [["false:false", "rule:true"]],\n            "example:early_or_success": [["rule:true"], ["false:false"]],\n            "example:lowercase_admin": [["role:admin"], ["role:sysadmin"]],\n            "example:uppercase_admin": [["role:ADMIN"], ["role:sysadmin"]],\n        }\n\n        # NOTE(vish): then overload underlying policy engine\n        self._set_rules()\n        self.credentials = {}\n        self.target = {}\n\n    def _set_rules(self):\n        these_rules = common_policy.Rules.from_dict(self.rules)\n        policy._ENFORCER._enforcer.set_rules(these_rules)\n\n    def test_enforce_nonexistent_action_throws(self):\n        action = "example:noexist"\n        self.assertRaises(exception.ForbiddenAction, policy.enforce,\n                          self.credentials, action, self.target)\n\n    def test_enforce_bad_action_throws(self):\n        action = "example:denied"\n        self.assertRaises(exception.ForbiddenAction, policy.enforce,\n                          self.credentials, action, self.target)\n\n    def test_enforce_good_action(self):\n        action = "example:allowed"\n        policy.enforce(self.credentials, action, self.target)\n\n    def test_templatized_enforcement(self):\n        target_mine = {\'project_id\': \'fake\'}\n        target_not_mine = {\'project_id\': \'another\'}\n        credentials = {\'project_id\': \'fake\', \'roles\': []}\n        action = "example:my_file"\n        policy.enforce(credentials, action, target_mine)\n        self.assertRaises(exception.ForbiddenAction, policy.enforce,\n                          credentials, action, target_not_mine)\n\n    def test_early_AND_enforcement(self):\n        action = "example:early_and_fail"\n        self.assertRaises(exception.ForbiddenAction, policy.enforce,\n                          self.credentials, action, self.target)\n\n    def test_early_OR_enforcement(self):\n        action = "example:early_or_success"\n        policy.enforce(self.credentials, action, self.target)\n\n    def test_ignore_case_role_check(self):\n        lowercase_action = "example:lowercase_admin"\n        uppercase_action = "example:uppercase_admin"\n        # NOTE(dprince): We mix case in the Admin role here to ensure\n        # case is ignored\n        admin_credentials = {\'roles\': [\'AdMiN\']}\n        policy.enforce(admin_credentials, lowercase_action, self.target)\n        policy.enforce(admin_credentials, uppercase_action, self.target)\n\n\nclass PolicyScopeTypesEnforcementTestCase(unit.TestCase):\n\n    def setUp(self):\n        super(PolicyScopeTypesEnforcementTestCase, self).setUp()\n        rule = common_policy.RuleDefault(\n            name=\'foo\',\n            check_str=\'\',\n            scope_types=[\'system\']\n        )\n        policy._ENFORCER._enforcer.register_default(rule)\n        self.credentials = {}\n        self.action = \'foo\'\n        self.target = {}\n\n    def test_forbidden_is_raised_if_enforce_scope_is_true(self):\n        self.config_fixture.config(group=\'oslo_policy\', enforce_scope=True)\n        self.assertRaises(\n            exception.ForbiddenAction, policy.enforce, self.credentials,\n            self.action, self.target\n        )\n\n    def test_warning_message_is_logged_if_enforce_scope_is_false(self):\n        self.config_fixture.config(group=\'oslo_policy\', enforce_scope=False)\n        expected_msg = (\n            \'Policy "foo": "" failed scope check. The token used to make the \'\n            \'request was project scoped but the policy requires [\\\'system\\\'] \'\n            \'scope. This behavior may change in the future where using the \'\n            \'intended scope is required\'\n        )\n        with mock.patch(\'warnings.warn\') as mock_warn:\n            policy.enforce(self.credentials, self.action, self.target)\n            mock_warn.assert_called_with(expected_msg)\n\n\nclass PolicyJsonTestCase(unit.TestCase):\n\n    def _get_default_policy_rules(self):\n        """Return a dictionary of all in-code policies.\n\n        All policies have a default value that is maintained in code.\n        This method returns a dictionary containing all default policies.\n        """\n        rules = dict()\n        for rule in policies.list_rules():\n            rules[rule.name] = rule.check_str\n        return rules\n\n    def test_policies_loads(self):\n        action = \'identity:list_projects\'\n        target = {\'user_id\': uuid.uuid4().hex,\n                  \'user.domain_id\': uuid.uuid4().hex,\n                  \'group.domain_id\': uuid.uuid4().hex,\n                  \'project.domain_id\': uuid.uuid4().hex,\n                  \'project_id\': uuid.uuid4().hex,\n                  \'domain_id\': uuid.uuid4().hex}\n        credentials = {\'username\': uuid.uuid4().hex, \'token\': uuid.uuid4().hex,\n                       \'project_name\': None, \'user_id\': uuid.uuid4().hex,\n                       \'roles\': [u\'admin\'], \'is_admin\': True,\n        self._old_load_rules = rules._ENFORCER.load_rules\n        self.addCleanup(setattr, rules._ENFORCER, \'load_rules\',\n                        self._old_load_rules)\n        setattr(rules._ENFORCER, \'load_rules\', lambda *args, **kwargs: None)\n\n    def _set_rules(self, default_rule):\n        these_rules = common_policy.Rules.from_dict(self.rules, default_rule)\n        self.assertTrue(result)\n\n    def test_all_targets_documented(self):\n        policy_keys = self._get_default_policy_rules()\n\n        # These keys are in the policy.yaml but aren\'t targets.\n        policy_rule_keys = [\n            \'admin_or_owner\', \'admin_or_token_subject\', \'admin_required\',\n            \'owner\', \'service_admin_or_token_subject\', \'service_or_admin\',\n            \'service_role\', \'token_subject\', ]\n\n        def read_doc_targets():\n            # Parse the doc/source/policy_mapping.rst file and return the\n            # targets.\n\n            doc_path = os.path.join(\n                unit.ROOTDIR, \'doc\', \'source\', \'getting-started\',\n                \'policy_mapping.rst\')\n            with open(doc_path) as doc_file:\n                for line in doc_file:\n                    if line.startswith(\'Target\'):\n                        break\n                for line in doc_file:\n                    # Skip === line\n                    if line.startswith(\'===\'):\n                        break\n                for line in doc_file:\n                    line = line.rstrip()\n                    if not line or line.startswith(\' \'):\n                        continue\n                    if line.startswith(\'==\'):\n                        break\n                    target, dummy, dummy = line.partition(\' \')\n                    yield str(target)\n\n        doc_targets = list(read_doc_targets())\n        self.assertCountEqual(policy_keys, doc_targets + policy_rule_keys)\n\n\nclass GeneratePolicyFileTestCase(unit.TestCase):\n\n    def test_policy_generator_from_command_line(self):\n        # This test ensures keystone.common.policy:get_enforcer ignores\n        # unexpected arguments before handing them off to oslo.config, which\n        # will fail and prevent users from generating policy files.\n        ret_val = subprocess.Popen(\n            [\'oslopolicy-policy-generator\', \'--namespace\', \'keystone\'],\n            stdout=subprocess.PIPE,\n            stderr=subprocess.PIPE\n        )\n        output = ret_val.communicate()\n        self.assertEqual(ret_val.returncode, 0, output)\n', 'message': 'Merge "Prefer . to setattr()/getattr()"', 'author_name': 'Gerrit Code Review', 'lang': 'py'}