def test_register_from_pack_action_metadata_fails_validation(self):
        # No fail on failure flag, should succeed
        pack_dir = os.path.join(get_fixtures_packs_base_path(), 'dummy_pack_4')
        runner_dirs = os.path.join(get_fixtures_packs_base_path(), 'runners')

        opts = [
            '--register-pack=%s' % (pack_dir),
            '--register-no-fail-on-failure',
            '--register-runner-dir=%s' % (runner_dirs),
        ]

        cmd = BASE_REGISTER_ACTIONS_CMD_ARGS + opts
        exit_code, _, stderr = run_command(cmd=cmd)
        self.assertTrue('Registered 0 actions.' in stderr)
        self.assertEqual(exit_code, 0)

        # Fail on failure, should fail
        pack_dir = os.path.join(get_fixtures_packs_base_path(), 'dummy_pack_4')
        opts = [
            '--register-pack=%s' % (pack_dir),
            '--register-fail-on-failure',
            '--register-runner-dir=%s' % (runner_dirs),
        ]
        cmd = BASE_REGISTER_ACTIONS_CMD_ARGS + opts
        exit_code, _, stderr = run_command(cmd=cmd)
        self.assertTrue('object has no attribute \'get\'' in stderr)
        self.assertEqual(exit_code, 1)

    def test_get_pack_resource_file_abs_path(self):
        # Mock the packs path to point to the fixtures directory
        cfg.CONF.content.packs_base_paths = get_fixtures_packs_base_path()

        # Invalid resource type
        expected_msg = 'Invalid resource type: fooo'
        self.assertRaisesRegexp(ValueError, expected_msg, get_pack_resource_file_abs_path,
                                pack_ref='dummy_pack_1',
                                resource_type='fooo',
                                file_path='test.py')

        # Invalid paths (directory traversal and absolute paths)
        file_paths = ['/tmp/foo.py', '../foo.py', '/etc/passwd', '../../foo.py']
        for file_path in file_paths:
            expected_msg = 'Invalid file path: .*%s' % (file_path)
            self.assertRaisesRegexp(ValueError, expected_msg, get_pack_resource_file_abs_path,
                                    pack_ref='dummy_pack_1',
                                    resource_type='action',
                                    file_path=file_path)

        # Valid paths
        file_paths = ['foo.py', 'a/foo.py', 'a/b/foo.py']
        for file_path in file_paths:
            expected = os.path.join(get_fixtures_packs_base_path(),
                                    'dummy_pack_1/actions', file_path)
            result = get_pack_resource_file_abs_path(pack_ref='dummy_pack_1',
                                                     resource_type='action',
                                                     file_path=file_path)
            self.assertEqual(result, expected)

    def test_register_from_pack_success(self):
        pack_dir = os.path.join(get_fixtures_packs_base_path(), 'dummy_pack_1')
        runner_dirs = os.path.join(get_fixtures_packs_base_path(), 'runners')

        opts = [
            '--register-pack=%s' % (pack_dir),
            '--register-runner-dir=%s' % (runner_dirs),
        ]
        cmd = BASE_REGISTER_ACTIONS_CMD_ARGS + opts
        exit_code, _, stderr = run_command(cmd=cmd)
        self.assertTrue('Registered 1 actions.' in stderr)
        self.assertEqual(exit_code, 0)

    def test_get_pack_resource_file_abs_path(self):
        # Mock the packs path to point to the fixtures directory
        cfg.CONF.content.packs_base_paths = get_fixtures_packs_base_path()

        # Invalid resource type
        expected_msg = 'Invalid resource type: fooo'
        self.assertRaisesRegexp(ValueError, expected_msg, get_pack_resource_file_abs_path,
                                pack_ref='dummy_pack_1',
                                resource_type='fooo',
                                file_path='test.py')

        # Invalid paths (directory traversal and absolute paths)
        file_paths = ['/tmp/foo.py', '../foo.py', '/etc/passwd', '../../foo.py',
                     '/opt/stackstorm/packs/invalid_pack/actions/my_action.py',
                     '../../foo.py']
        for file_path in file_paths:
            # action resource_type
            expected_msg = (r'Invalid file path: ".*%s"\. File path needs to be relative to the '
                            r'pack actions directory (.*). For example "my_action.py"\.' %
                            (file_path))
            self.assertRaisesRegexp(ValueError, expected_msg, get_pack_resource_file_abs_path,
                                    pack_ref='dummy_pack_1',
                                    resource_type='action',
                                    file_path=file_path)

            # sensor resource_type
            expected_msg = (r'Invalid file path: ".*%s"\. File path needs to be relative to the '
                            r'pack sensors directory (.*). For example "my_sensor.py"\.' %
                            (file_path))
            self.assertRaisesRegexp(ValueError, expected_msg, get_pack_resource_file_abs_path,
                                    pack_ref='dummy_pack_1',
                                    resource_type='sensor',
                                    file_path=file_path)

            # no resource type
            expected_msg = (r'Invalid file path: ".*%s"\. File path needs to be relative to the '
                            r'pack directory (.*). For example "my_action.py"\.' %
                            (file_path))
            self.assertRaisesRegexp(ValueError, expected_msg, get_pack_file_abs_path,
                                    pack_ref='dummy_pack_1',
                                    file_path=file_path)

        # Valid paths
        file_paths = ['foo.py', 'a/foo.py', 'a/b/foo.py']
        for file_path in file_paths:
            expected = os.path.join(get_fixtures_packs_base_path(),
                                    'dummy_pack_1/actions', file_path)
            result = get_pack_resource_file_abs_path(pack_ref='dummy_pack_1',
                                                     resource_type='action',
                                                     file_path=file_path)
            self.assertEqual(result, expected)

    def test_register_triggers_from_pack(self):
        base_path = get_fixtures_packs_base_path()
        pack_dir = os.path.join(base_path, 'dummy_pack_1')

        trigger_type_dbs = TriggerType.get_all()
        self.assertEqual(len(trigger_type_dbs), 0)

        count = triggers_registrar.register_triggers(pack_dir=pack_dir)
        self.assertEqual(count, 2)

        # Verify TriggerTypeDB and corresponding TriggerDB objects have been created
        trigger_type_dbs = TriggerType.get_all()
        trigger_dbs = Trigger.get_all()
        self.assertEqual(len(trigger_type_dbs), 2)
        self.assertEqual(len(trigger_dbs), 2)

        self.assertEqual(trigger_type_dbs[0].name, 'event_handler')
        self.assertEqual(trigger_type_dbs[0].pack, 'dummy_pack_1')
        self.assertEqual(trigger_dbs[0].name, 'event_handler')
        self.assertEqual(trigger_dbs[0].pack, 'dummy_pack_1')
        self.assertEqual(trigger_dbs[0].type, 'dummy_pack_1.event_handler')

        self.assertEqual(trigger_type_dbs[1].name, 'head_sha_monitor')
        self.assertEqual(trigger_type_dbs[1].pack, 'dummy_pack_1')
        self.assertEqual(trigger_type_dbs[1].payload_schema['type'], 'object')

    def test_register_all_policies(self):
        policies_dbs = Policy.get_all()
        self.assertEqual(len(policies_dbs), 0)

        packs_base_path = get_fixtures_packs_base_path()
        count = policies_registrar.register_policies(packs_base_paths=[packs_base_path])

        # Verify PolicyDB objects have been created
        policies_dbs = Policy.get_all()

        policies = {
            policies_db.name: {
                'pack': policies_db.pack,
                'type': policies_db.policy_type,
                'parameters': policies_db.parameters
            }
            for policies_db in policies_dbs
        }

        expected_policies = {
            'test_policy_1': {
                'pack': 'dummy_pack_1',
                'type': 'action.concurrency',
                'parameters': {
                    'action': 'delay',
                    'threshold': 3
                }
            },
            'test_policy_3': {
                'pack': 'dummy_pack_1',
                'type': 'action.retry',
                'parameters': {
                    'retry_on': 'timeout',
                    'max_retry_count': 5
                }
            },
            'cancel_on_concurrency': {
                'pack': 'mistral_tests',
                'type': 'action.concurrency',
                'parameters': {
                    'action': 'cancel',
                    'threshold': 3
                }
            },
            'cancel_on_concurrency_by_attr': {
                'pack': 'mistral_tests',
                'type': 'action.concurrency.attr',
                'parameters': {
                    'action': 'cancel',
                    'threshold': 1,
                    'attributes': ['friend']
                }
            }
        }

        self.assertEqual(len(expected_policies), count)
        self.assertEqual(len(expected_policies), len(policies_dbs))
        self.assertDictEqual(expected_policies, policies)

def _override_common_opts():
    packs_base_path = get_fixtures_packs_base_path()
    CONF.set_override(name='base_path', override=packs_base_path, group='system')
    CONF.set_override(name='system_packs_base_path', override=packs_base_path, group='content')
    CONF.set_override(name='packs_base_paths', override=packs_base_path, group='content')
    CONF.set_override(name='api_url', override='http://127.0.0.1', group='auth')
    CONF.set_override(name='mask_secrets', override=True, group='log')
    CONF.set_override(name='url', override='zake://', group='coordination')
    CONF.set_override(name='lock_timeout', override=1, group='coordination')

    def test_register_policy_invalid_policy_type_references(self):
        # Policy references an invalid (inexistent) policy type
        registrar = PolicyRegistrar()
        policy_path = os.path.join(get_fixtures_packs_base_path(),
                                   'dummy_pack_1/policies/policy_2.yaml')

        expected_msg = 'Referenced policy_type "action.mock_policy_error" doesnt exist'
        self.assertRaisesRegexp(ValueError, expected_msg, registrar._register_policy,
                                pack='dummy_pack_1', policy=policy_path)

    def test_make_sure_policy_parameters_are_validated_during_register(self):
        # Policy where specified parameters fail schema validation
        registrar = PolicyRegistrar()
        policy_path = os.path.join(get_fixtures_packs_base_path(),
                                   'dummy_pack_2/policies/policy_3.yaml')

        expected_msg = '100 is greater than the maximum of 5'
        self.assertRaisesRegexp(jsonschema.ValidationError, expected_msg,
                                registrar._register_policy, pack='dummy_pack_2',
                                policy=policy_path)

    def test_register_setup_virtualenvs(self):
        # Single pack
        pack_dir = os.path.join(get_fixtures_packs_base_path(), 'dummy_pack_1')

        cmd = BASE_CMD_ARGS + ['--register-pack=%s' % (pack_dir), '--register-setup-virtualenvs',
                               '--register-no-fail-on-failure']
        exit_code, stdout, stderr = run_command(cmd=cmd)

        self.assertTrue('Setting up virtualenv for pack "dummy_pack_1"' in stderr)
        self.assertTrue('Setup virtualenv for 1 pack(s)' in stderr)
        self.assertEqual(exit_code, 0)

def _override_common_opts():
    packs_base_path = get_fixtures_packs_base_path()
    CONF.set_override(name='base_path', override=packs_base_path, group='system')
    CONF.set_override(name='validate_output_schema', override=True, group='system')
    CONF.set_override(name='system_packs_base_path', override=packs_base_path, group='content')
    CONF.set_override(name='packs_base_paths', override=packs_base_path, group='content')
    CONF.set_override(name='api_url', override='http://127.0.0.1', group='auth')
    CONF.set_override(name='mask_secrets', override=True, group='log')
    CONF.set_override(name='jitter_interval', override=0, group='mistral')
    CONF.set_override(name='query_interval', override=0.1, group='resultstracker')
    CONF.set_override(name='stream_output', override=False, group='actionrunner')

    def test_register_all_triggers(self):
        trigger_type_dbs = TriggerType.get_all()
        self.assertEqual(len(trigger_type_dbs), 0)

        packs_base_path = get_fixtures_packs_base_path()
        count = triggers_registrar.register_triggers(packs_base_paths=[packs_base_path])
        self.assertEqual(count, 2)

        # Verify TriggerTypeDB and corresponding TriggerDB objects have been created
        trigger_type_dbs = TriggerType.get_all()
        trigger_dbs = Trigger.get_all()
        self.assertEqual(len(trigger_type_dbs), 2)
        self.assertEqual(len(trigger_dbs), 2)

    def test_register_policies(self):
        # Note: Only one policy should be registered since second one fails validation
        pack_dir = os.path.join(fixturesloader.get_fixtures_packs_base_path(), 'dummy_pack_1')
        self.assertEqual(register_policies(pack_dir=pack_dir), 1)

        p1 = Policy.get_by_ref('dummy_pack_1.test_policy_1')
        self.assertEqual(p1.name, 'test_policy_1')
        self.assertEqual(p1.pack, 'dummy_pack_1')
        self.assertEqual(p1.resource_ref, 'dummy_pack_1.local')
        self.assertEqual(p1.policy_type, 'action.concurrency')

        p2 = Policy.get_by_ref('dummy_pack_1.test_policy_2')
        self.assertEqual(p2, None)

    def test_register_policies_from_pack(self):
        pack_dir = os.path.join(get_fixtures_packs_base_path(), 'dummy_pack_1')
        self.assertEqual(register_policies(pack_dir=pack_dir), 2)

        p1 = Policy.get_by_ref('dummy_pack_1.test_policy_1')
        self.assertEqual(p1.name, 'test_policy_1')
        self.assertEqual(p1.pack, 'dummy_pack_1')
        self.assertEqual(p1.resource_ref, 'dummy_pack_1.local')
        self.assertEqual(p1.policy_type, 'action.concurrency')
        # Verify that a default value for parameter "action" which isn't provided in the file is set
        self.assertEqual(p1.parameters['action'], 'delay')

        p2 = Policy.get_by_ref('dummy_pack_1.test_policy_2')
        self.assertEqual(p2, None)

 def test_register_all_and_register_setup_virtualenvs(self):
     # Verify that --register-all works in combinations with --register-setup-virtualenvs
     # Single pack
     pack_dir = os.path.join(get_fixtures_packs_base_path(), 'dummy_pack_1')
     cmd = BASE_CMD_ARGS + [
         '--register-pack=%s' % (pack_dir),
         '--register-all',
         '--register-setup-virtualenvs',
         '--register-no-fail-on-failure'
     ]
     exit_code, stdout, stderr = run_command(cmd=cmd)
     self.assertTrue('Registering actions' in stderr, 'Actual stderr: %s' % (stderr))
     self.assertTrue('Registering rules' in stderr)
     self.assertTrue('Setup virtualenv for %s pack(s)' % ('1') in stderr)
     self.assertEqual(exit_code, 0)

    def test_get_pack_resource_file_abs_path(self):
        # Mock the packs path to point to the fixtures directory
        cfg.CONF.content.packs_base_paths = get_fixtures_packs_base_path()

        # Invalid resource type
        expected_msg = "Invalid resource type: fooo"
        self.assertRaisesRegexp(
            ValueError,
            expected_msg,
            get_pack_resource_file_abs_path,
            pack_name="dummy_pack_1",
            resource_type="fooo",
            file_path="test.py",
        )

        # Invalid paths (directory traversal and absolute paths)
        file_paths = ["/tmp/foo.py", "../foo.py", "/etc/passwd", "../../foo.py"]
        for file_path in file_paths:
            expected_msg = "Invalid file path: .*%s" % (file_path)
            self.assertRaisesRegexp(
                ValueError,
                expected_msg,
                get_pack_resource_file_abs_path,
                pack_name="dummy_pack_1",
                resource_type="action",
                file_path=file_path,
            )

        # Valid paths
        file_paths = ["foo.py", "a/foo.py", "a/b/foo.py"]
        for file_path in file_paths:
            expected = os.path.join(get_fixtures_packs_base_path(), "dummy_pack_1/actions", file_path)
            result = get_pack_resource_file_abs_path(
                pack_name="dummy_pack_1", resource_type="action", file_path=file_path
            )
            self.assertEqual(result, expected)

    def test_register_all_policies(self):
        policies_dbs = Policy.get_all()
        self.assertEqual(len(policies_dbs), 0)

        packs_base_path = get_fixtures_packs_base_path()
        count = policies_registrar.register_policies(packs_base_paths=[packs_base_path])
        self.assertEqual(count, 2)

        # Verify PolicyDB objects have been created
        policies_dbs = Policy.get_all()
        self.assertEqual(len(policies_dbs), 2)

        self.assertEqual(policies_dbs[0].name, 'test_policy_1')
        self.assertEqual(policies_dbs[0].policy_type, 'action.concurrency')

        # Verify that a default value for parameter "action" which isn't provided in the file is set
        self.assertEqual(policies_dbs[0].parameters['action'], 'delay')
        self.assertEqual(policies_dbs[0].parameters['threshold'], 3)

    def test_get_relative_path_to_pack_file(self):
        packs_base_paths = get_fixtures_packs_base_path()

        pack_ref = 'dummy_pack_1'

        # 1. Valid paths
        file_path = os.path.join(packs_base_paths, 'dummy_pack_1/pack.yaml')
        result = get_relative_path_to_pack_file(pack_ref=pack_ref, file_path=file_path)
        self.assertEqual(result, 'pack.yaml')

        file_path = os.path.join(packs_base_paths, 'dummy_pack_1/actions/action.meta.yaml')
        result = get_relative_path_to_pack_file(pack_ref=pack_ref, file_path=file_path)
        self.assertEqual(result, 'actions/action.meta.yaml')

        file_path = os.path.join(packs_base_paths, 'dummy_pack_1/actions/lib/foo.py')
        result = get_relative_path_to_pack_file(pack_ref=pack_ref, file_path=file_path)
        self.assertEqual(result, 'actions/lib/foo.py')

        # Already relative
        file_path = 'actions/lib/foo2.py'
        result = get_relative_path_to_pack_file(pack_ref=pack_ref, file_path=file_path)
        self.assertEqual(result, 'actions/lib/foo2.py')

        # 2. Invalid path - outside pack directory
        expected_msg = r'file_path (.*?) is not located inside the pack directory (.*?)'

        file_path = os.path.join(packs_base_paths, 'dummy_pack_2/actions/lib/foo.py')
        self.assertRaisesRegexp(ValueError, expected_msg, get_relative_path_to_pack_file,
                                pack_ref=pack_ref, file_path=file_path)

        file_path = '/tmp/foo/bar.py'
        self.assertRaisesRegexp(ValueError, expected_msg, get_relative_path_to_pack_file,
                                pack_ref=pack_ref, file_path=file_path)

        file_path = os.path.join(packs_base_paths, '../dummy_pack_1/pack.yaml')
        self.assertRaisesRegexp(ValueError, expected_msg, get_relative_path_to_pack_file,
                                pack_ref=pack_ref, file_path=file_path)

        file_path = os.path.join(packs_base_paths, '../../dummy_pack_1/pack.yaml')
        self.assertRaisesRegexp(ValueError, expected_msg, get_relative_path_to_pack_file,
                                pack_ref=pack_ref, file_path=file_path)

    def test_register_from_pack_fail_on_failure_pack_dir_doesnt_exist(self):
        # No fail on failure flag, should succeed
        pack_dir = 'doesntexistblah'
        runner_dirs = os.path.join(get_fixtures_packs_base_path(), 'runners')

        opts = [
            '--register-pack=%s' % (pack_dir),
            '--register-runner-dir=%s' % (runner_dirs),
            '--register-no-fail-on-failure'
        ]
        cmd = BASE_REGISTER_ACTIONS_CMD_ARGS + opts
        exit_code, _, _ = run_command(cmd=cmd)
        self.assertEqual(exit_code, 0)

        # Fail on failure, should fail
        opts = [
            '--register-pack=%s' % (pack_dir),
            '--register-runner-dir=%s' % (runner_dirs),
            '--register-fail-on-failure'
        ]
        cmd = BASE_REGISTER_ACTIONS_CMD_ARGS + opts
        exit_code, _, stderr = run_command(cmd=cmd)
        self.assertTrue('Directory "doesntexistblah" doesn\'t exist' in stderr)
        self.assertEqual(exit_code, 1)

from st2common.persistence.rule import Rule
from st2common.persistence.policy import Policy
from st2common.persistence.sensor import SensorType as Sensor
from st2common.persistence.trigger import TriggerType

from st2tests.base import BaseActionTestCase
from st2tests.base import CleanDbTestCase
from st2tests import fixturesloader

from pack_mgmt.unload import UnregisterPackAction

__all__ = [
    'UnloadActionTestCase'
]

PACK_PATH_1 = os.path.join(fixturesloader.get_fixtures_packs_base_path(), 'dummy_pack_1')


class UnloadActionTestCase(BaseActionTestCase, CleanDbTestCase):
    action_cls = UnregisterPackAction

    def setUp(self):
        super(UnloadActionTestCase, self).setUp()

        # Register mock pack
        # Verify DB is empty
        pack_dbs = Pack.get_all()
        config_schema_dbs = ConfigSchema.get_all()
        config_dbs = Config.get_all()

        self.assertEqual(len(pack_dbs), 0)