~jcsackett/charmworld/bac-tag-constraints

# Copyright 2012, 2013 Marco Ceppi, Canonical Ltd.  This software is
# licensed under the GNU Affero General Public License version 3 (see
# the file LICENSE).

"""Unit tests for our application models."""
from datetime import (
    date,
    datetime,
)
import logging
import md5
import os
from os.path import dirname
from os.path import join
from os.path import split

from mock import patch
from pyelasticsearch.exceptions import (
    ElasticHttpError,
    ElasticHttpNotFoundError,
)

from charmworld.models import (
    acquire_session_secret,
    Bundle,
    Charm,
    CharmFile,
    CharmFileSet,
    CharmSource,
    find_charms,
    options_to_storage,
    QAData,
    storage_to_options,
    sync_index,
    UserMgr,
)
from charmworld.testing import (
    factory,
    force_index_client,
    INI,
    MongoTestBase,
    temp_dir,
    TestCase,
)
from charmworld.testing.factory import charm_branch
from charmworld.utils import (
    quote_key,
)


class CharmTestCase(TestCase):
    """Unit tests for the Charm model that wraps charm representations."""

    def test_init_with_invalid_charm(self):
        charm_data = {}
        charm = Charm(charm_data)
        self.assertIs(charm_data, charm._raw_representation)
        self.assertEqual(charm._COMMON_REPRESENTATION, charm._representation)

    def test_init_with_valid_charm(self):
        charm_data = factory.get_charm_json(
            name='name', owner='owner', series='series', promulgated=True)
        charm = Charm(charm_data)
        self.assertIs('name', charm._representation['name'])
        self.assertIs('owner', charm._representation['owner'])
        self.assertIs('series', charm._representation['series'])
        self.assertIs(True, charm._representation['promulgated'])

    def test_eq(self):
        charm_1 = Charm(factory.get_charm_json(
            owner='owner', series='series', name='name'))
        charm_2 = Charm(factory.get_charm_json(
            owner='other', series='series', name='name'))
        charm_3 = Charm(factory.get_charm_json(
            owner=charm_1.owner, series=charm_1.series, name=charm_1.name))
        self.assertNotEqual(charm_1, charm_2)
        self.assertEqual(charm_1, charm_3)

    def test_branch_spec(self):
        # The branch_spec property is a string.
        charm_data = factory.get_charm_json(
            owner='owner', series='series', name='name')
        charm = Charm(charm_data)
        self.assertEqual(
            '~owner/charms/series/name/trunk', charm.branch_spec)

    def test_bzr_branch(self):
        # The bzr_branch property is a string.
        charm_data = factory.get_charm_json(
            owner='owner', series='series', name='name')
        charm = Charm(charm_data)
        self.assertEqual(
            'lp:~owner/charms/series/name/trunk', charm.bzr_branch)

    def test_owner(self):
        # The owner property is a string.
        charm_data = factory.get_charm_json(owner='owner')
        charm = Charm(charm_data)
        self.assertEqual('owner', charm.owner)

    def test_series(self):
        # The series property is a string.
        charm_data = factory.get_charm_json(series='series')
        charm = Charm(charm_data)
        self.assertEqual('series', charm.series)

    def test_name(self):
        # The name property is a string.
        charm_data = factory.get_charm_json(name='name')
        charm = Charm(charm_data)
        self.assertEqual('name', charm.name)

    def test_bname(self):
        # The bname property is a string.
        charm_data = factory.get_charm_json(bname='trunk')
        charm = Charm(charm_data)
        self.assertEqual('trunk', charm.bname)

    def test_commit(self):
        # The commit property is a revid string.
        charm_data = factory.get_charm_json()
        charm_data['commit'] = 'joe@example.com-20120723223205-babble'
        charm = Charm(charm_data)
        self.assertEqual(
            'joe@example.com-20120723223205-babble', charm.commit)
        # The default is an empty string that happens when a branch is empty.
        charm = Charm({})
        self.assertEqual('', charm.commit)

    def test_distro_series(self):
        # The distro_series property is a list of series names.
        charm_data = factory.get_charm_json()
        charm_data['distro_series'] = distro_series = ['series']
        charm = Charm(charm_data)
        self.assertIs(distro_series, charm.distro_series)
        # The default is an empty list.
        charm = Charm({})
        self.assertEqual([], charm.distro_series)

    def test_promulgated(self):
        # The promulgated property is a bool.
        charm_data = factory.get_charm_json(promulgated=True)
        charm = Charm(charm_data)
        self.assertIs(True, charm.promulgated)
        # The default is False.
        charm = Charm({})
        self.assertIs(False, charm.promulgated)

    def test_id(self):
        # The _id property is string string based on the branch_spec.
        charm_data = factory.get_charm_json(
            owner='owner', series='series', name='name')
        charm_data['_id'] = '~owner/charms/series/name/trunk'
        charm = Charm(charm_data)
        self.assertEqual('~owner/charms/series/name/trunk', charm._id)
        # The _id property is actually arbitrary.
        charm = Charm({'_id': 'boing'})
        self.assertEqual('boing', charm._id)

    def test_date_created(self):
        # The date_created property is an ISO timestamp.
        date_created = factory.random_date()
        charm_data = factory.get_charm_json(date_created=date_created)
        charm = Charm(charm_data)
        self.assertEqual(date_created.isoformat(), charm.date_created)
        # The default is None, which happens when the branch cannot be found.
        charm = Charm({})
        self.assertEquals(None, charm.date_created)

    def test_error(self):
        # The error property is a string.
        charm_data = {'error': 'Paris is burning'}
        charm = Charm(charm_data)
        self.assertEqual('Paris is burning', charm.error)
        # The default is an empty string.
        charm = Charm({})
        self.assertEquals('', charm.error)

    def test_branch_dir(self):
        # The branch_dir property is a string.
        charm_data = {'branch_dir': '/var/charms/series/owner/name/bname'}
        charm = Charm(charm_data)
        self.assertEqual(
            '/var/charms/series/owner/name/bname', charm.branch_dir)
        # The default is an empty string.
        charm = Charm({})
        self.assertEquals('', charm.branch_dir)

    def test_files(self):
        # The files property is a dict.
        files = {
            'install': {
                'filename': 'install',
                'subdir': 'hooks',
            },
        }
        charm_data = factory.get_charm_json(files=files)
        charm = Charm(charm_data)
        self.assertIs(files, charm.files)
        # The default is an empty string.
        charm = Charm({})
        self.assertEquals({}, charm.files)

    def test_downloads(self):
        """The downloads property is an int."""
        charm_data = {'downloads': 30}
        charm = Charm(charm_data)
        self.assertEqual(30, charm.downloads)
        # The default is zero.
        charm = Charm({})
        self.assertEquals(0, charm.downloads)

    def test_downloads_in_past_30_days(self):
        # The downloads_in_past_30_days property is an int.
        charm_data = {'downloads_in_past_30_days': 30}
        charm = Charm(charm_data)
        self.assertEqual(30, charm.downloads_in_past_30_days)
        # The default is zero.
        charm = Charm({})
        self.assertEquals(0, charm.downloads_in_past_30_days)

    def test_store_url(self):
        # The store_url property is a string.
        charm_data = {'store_url': 'cs:~owner/series/name-1'}
        charm = Charm(charm_data)
        self.assertEqual('cs:~owner/series/name-1', charm.store_url)
        # The default is an empty sting..
        charm = Charm({})
        self.assertEquals('', charm.store_url)

    def test_store_revision(self):
        # The store_revision property is an int.
        charm_data = {'store_data': {'revision': 17}}
        charm = Charm(charm_data)
        self.assertEqual(17, charm.store_revision)
        # The default is 1 if revision is missing from store_data.
        charm = Charm({'store_data': {}})
        self.assertEquals(1, charm.store_revision)
        # The default is 1 if there is no store_data.
        charm = Charm({})
        self.assertEquals(1, charm.store_revision)

    def test_store_data(self):
        # The store_data property dict.
        charm_data = factory.get_charm_json()
        store_data = {
            'address': 'cs:~owner/series/name-1',
            'store_checked': 'Thu Dec  6 17:42:05 2012',
            'digest': 'jdoe@example.com-20120723223205-babble',
            'revision': 1,
            'sha256': '107c230dc3481f8b8de9decfff6439a23caae2ea670bddb413ddf1',
        }
        charm_data['store_data'] = store_data
        charm = Charm(charm_data)
        self.assertIs(store_data, charm.store_data)
        # The default is an empty dict.
        charm = Charm({})
        self.assertEqual({}, charm.store_data)

    def test_proof(self):
        # The proof property is a dict.
        proof = {
            'w': [' no README file']
        }
        charm_data = factory.get_charm_json()
        charm_data['proof'] = proof
        charm = Charm(charm_data)
        self.assertIs(proof, charm.proof)
        # The default is an empty dict.
        charm = Charm({})
        self.assertEqual({}, charm.proof)

    def test_tests(self):
        # The tests property is a dict.
        tests = {
            'ec2': 'PASS'
        }
        charm_data = factory.get_charm_json()
        charm_data['tests'] = tests
        charm = Charm(charm_data)
        self.assertIs(tests, charm.tests)
        # The default is an empty dict.
        charm = Charm({})
        self.assertEqual({}, charm.tests)

    def test_test_results(self):
        # The test_results property is a dict.
        test_results = {
            'openstack': 987654321
        }
        charm_data = factory.get_charm_json()
        charm_data['test_results'] = test_results
        charm = Charm(charm_data)
        self.assertIs(test_results, charm.test_results)
        # The default is an empty dict.
        charm = Charm({})
        self.assertEqual({}, charm.test_results)

    def test_changes(self):
        # The changes property is a list of revision info dicts.
        changes = [factory.makeChange()]
        charm_data = factory.get_charm_json(changes=changes)
        charm = Charm(charm_data)
        self.assertIs(changes, charm.changes)
        # The default is an empty list.
        charm = Charm({})
        self.assertEqual([], charm.changes)

    def test_first_change(self):
        # The first_change property is a revision info dict.
        change = factory.makeChange()
        charm_data = factory.get_charm_json()
        charm_data['first_change'] = change
        charm = Charm(charm_data)
        self.assertIs(change, charm.first_change)
        # The default is None.
        charm = Charm({})
        self.assertIs(None, charm.first_change)

    def test_last_change(self):
        # The last_change property is a revision info dict.
        change = factory.makeChange()
        charm_data = factory.get_charm_json()
        charm_data['last_change'] = change
        charm = Charm(charm_data)
        self.assertIs(change, charm.last_change)
        # The default is None.
        charm = Charm({})
        self.assertIs(None, charm.last_change)

    def test_revision(self):
        # The revision property is an int.
        charm_data = factory.get_charm_json(revision=5)
        charm = Charm(charm_data)
        self.assertEqual(5, charm.revision)
        # The default is zero.
        charm = Charm({})
        self.assertEquals(0, charm.revision)

    def test_summary(self):
        # The summary property is string.
        summary = 'text'
        charm_data = factory.get_charm_json()
        charm_data['summary'] = summary
        charm = Charm(charm_data)
        self.assertIs(summary, charm.summary)
        # The default is an empty string.
        charm = Charm({})
        self.assertEqual('', charm.summary)

    def test_description(self):
        # The description property is string.
        description = 'text'
        charm_data = factory.get_charm_json()
        charm_data['description'] = description
        charm = Charm(charm_data)
        self.assertIs(description, charm.description)
        # The default is an empty string.
        charm = Charm({})
        self.assertEqual('', charm.description)

    def test_categories(self):
        # The categories property is a list of category names.
        categories = ['cache-proxy', 'app-servers']
        charm_data = factory.get_charm_json(categories=categories)
        charm = Charm(charm_data)
        self.assertEqual(categories, charm.categories)
        # The default is an empty list.
        charm = Charm({})
        self.assertEqual([], charm.categories)

    def test_categories_are_official(self):
        # Unofficial categories are filtered out of the list.
        categories = ['application']
        charm_data = factory.get_charm_json(categories=categories)
        charm = Charm(charm_data)
        self.assertEqual(
            ['applications', 'app-servers', 'cache-proxy', 'databases',
             'file-servers', 'misc'],
            charm.OFFICIAL_CATEGORIES)
        self.assertEqual([], charm.categories)

    def test_maintainer(self):
        # The maintainer property is a list of category names.
        maintainer = ['Jane <jane@example.com>', 'Ian <ian@example.com>']
        charm_data = factory.get_charm_json(maintainer=maintainer)
        charm = Charm(charm_data)
        self.assertIs(maintainer, charm.maintainer)
        # The default is an empty string.
        charm = Charm({})
        self.assertEqual('', charm.maintainer)

    def test_subordinate(self):
        # The subordinate property is a bool.
        charm_data = factory.get_charm_json(subordinate=True)
        charm = Charm(charm_data)
        self.assertIs(True, charm.subordinate)
        # The default is False.
        charm = Charm({})
        self.assertIs(False, charm.subordinate)

    @staticmethod
    def make_provides():
        return {
            "website": {
                "interface": "https",
            }
        }

    def test_provides(self):
        # The provides property is a dict of charm options.
        provides = self.make_provides()
        charm_data = factory.get_charm_json(provides=provides)
        charm = Charm(charm_data)
        self.assertIs(provides, charm.provides)
        # The default is an empty dict.
        charm = Charm({})
        self.assertEqual({}, charm.provides)

    def test_i_provides(self):
        # The i_provides property is a list of interface names.
        provides = self.make_provides()
        charm_data = factory.get_charm_json(provides=provides)
        charm = Charm(charm_data)
        self.assertIs(charm_data['i_provides'], charm.i_provides)
        # The default is an empty list.
        charm = Charm({})
        self.assertEqual([], charm.i_provides)

    @staticmethod
    def make_requires():
        return {
            'database': {
                'interface': 'mongodb',
            }
        }

    def test_requires(self):
        # The requires property is a dict of charm options.
        requires = self.make_requires()
        charm_data = factory.get_charm_json(requires=requires)
        charm = Charm(charm_data)
        self.assertIs(requires, charm.requires)
        # The default is an empty dict.
        charm = Charm({})
        self.assertEqual({}, charm.requires)

    def test_i_requires(self):
        # The i_provides property is a list of interface names.
        requires = self.make_requires()
        charm_data = factory.get_charm_json(requires=requires)
        charm = Charm(charm_data)
        self.assertIs(charm_data['i_requires'], charm.i_requires)
        # The default is an empty list.
        charm = Charm({})
        self.assertEqual([], charm.i_requires)

    def test_config(self):
        # The config property is a dict of charm options.
        config = {'options': {'key': {'default': 'value', 'type': 'string'}}}
        charm_data = factory.get_charm_json(options=config['options'])
        charm = Charm(charm_data)
        self.assertEqual(config, charm.config)
        # The default is a dict with a single key named 'options' set to
        # an empty dict.
        charm = Charm({})
        self.assertEqual({'options': {}}, charm.config)

    def test_config_raw(self):
        # The config_raw property is string of YAML.
        config_raw = 'option: 0'
        charm_data = factory.get_charm_json()
        charm_data['config_raw'] = config_raw
        charm = Charm(charm_data)
        self.assertIs(config_raw, charm.config_raw)
        # The default is an empty string.
        charm = Charm({})
        self.assertEqual('', charm.config_raw)

    def test_options(self):
        # The options property is a dict from the charm's config.
        options = {'key': {'default': 'value', 'type': 'string'}}
        charm_data = factory.get_charm_json(options=options)
        charm = Charm(charm_data)
        self.assertEqual(options, charm.options)
        # The default is a dict.
        charm = Charm({})
        self.assertEqual({}, charm.options)

    def test_options_when_config_is_none(self):
        # The options property is a dict even if the config is None.
        charm = Charm({})
        charm._representation['config'] = None
        self.assertEqual({}, charm.options)

    def test_hooks(self):
        # The hooks property is a list of life cycle and relation names.
        hooks = ['install', 'start', 'stop']
        charm_data = factory.get_charm_json()
        charm_data['hooks'] = hooks
        charm = Charm(charm_data)
        self.assertIs(hooks, charm.hooks)
        # The default is an empty list.
        charm = Charm({})
        self.assertEqual([], charm.hooks)

    def test_label(self):
        # The label property is string.
        label = 'user:series/name'
        charm_data = factory.get_charm_json()
        charm_data['label'] = label
        charm = Charm(charm_data)
        self.assertIs(label, charm.label)
        # The default is an empty string.
        charm = Charm({})
        self.assertEqual('', charm.label)

    def test_short_url(self):
        # The short_url property is string.
        short_url = 'user:series/name'
        charm_data = factory.get_charm_json()
        charm_data['short_url'] = short_url
        charm = Charm(charm_data)
        self.assertIs(short_url, charm.short_url)
        # The default is an empty string.
        charm = Charm({})
        self.assertEqual('', charm.short_url)

    def test_qa(self):
        # The qa property is a dict.
        qa_data = factory.makeQAScore()
        charm_data = factory.get_charm_json()
        charm_data['qa'] = qa_data
        charm = Charm(charm_data)
        self.assertIs(qa_data, charm.qa)
        # The default is None.
        charm = Charm({})
        self.assertIs(None, charm.qa)

    def test_is_featured(self):
        # The is_featured property is a bool.
        charm_data = factory.get_charm_json(is_featured=True)
        charm = Charm(charm_data)
        self.assertIs(True, charm.is_featured)
        # The default is False.
        charm = Charm({})
        self.assertIs(False, charm.is_featured)

    def test_changes_since(self):
        changes = [factory.makeChange(created=date(2013, 6, num))
                   for num in range(1, 31)]
        charm = Charm(factory.get_charm_json(changes=changes))
        self.assertEqual([], charm.changes_since(datetime(2013, 7, 1)))
        self.assertEqual(30, len(charm.changes_since(datetime(2013, 5, 1))))
        self.assertEqual(5, len(charm.changes_since(datetime(2013, 6, 26))))


class TestUser(MongoTestBase):
    """Unit tests for our User object wrapping of the mongo dict."""

    def test_model_from_sso(self):
        """Given an SSO dump we can get back a decent model object."""
        payload = factory.makeSSOResponse()
        user = UserMgr.from_sso(payload)

        self.assertEqual(payload['openid.claimed_id'], user.userid)
        self.assertEqual(payload['openid.sreg.nickname'], user.nickname)
        self.assertEqual(payload['openid.sreg.fullname'], user.fullname)
        self.assertEqual(payload['openid.sreg.email'], user.email)
        self.assertEqual(payload['openid.sreg.timezone'], user.timezone)
        self.assertEqual(payload['openid.lp.is_member'], ','.join(user.groups))

    def test_user_limited_sharing(self):
        """The user might limit information shared."""
        claimed_id = u'https://login.ubuntu.com/+id/MSbTCYM'

        user = UserMgr.from_sso({
            'openid.claimed_id': claimed_id,
            # When a user chooses not to share teams, an empty is_member is
            # returned.
            'openid.lp.is_member': '',
        })

        self.assertEqual(claimed_id, user.userid)
        self.assertEqual('Not Provided', user.nickname)
        self.assertEqual('Not Provided', user.fullname)
        self.assertEqual(None, user.email)
        self.assertEqual('GMT', user.timezone)
        self.assertEqual([], user.groups)

    def test_can_dict_user(self):
        """We need to be able to turn User into a dict for mongo."""
        payload = factory.makeSSOResponse()
        user = UserMgr.from_sso(payload)

        dict_user = dict(user)
        self.assertEqual(
            payload['openid.claimed_id'],
            dict_user['userid'])
        self.assertEqual(
            payload['openid.sreg.nickname'],
            dict_user['nickname'])

    def test_user_stored_in_mongo(self):
        """The user should store in mongo correctly on save."""
        payload = factory.makeSSOResponse()
        user = UserMgr.from_sso(payload)
        user.save(self.db)

        found = self.db.users.find_one({'userid': user.userid})
        self.assertEqual(user.userid, found['userid'])

    def test_user_from_mongo(self):
        """Pulling the user from mongo via UserMgr should get a User out."""
        payload = factory.makeSSOResponse()
        user = UserMgr.from_sso(payload)
        user.save(self.db)

        found = UserMgr.find_one(self.db, user.userid)
        self.assertEqual(user.userid, found.userid)

    def test_user_updated(self):
        """Saving will either insert or save as required."""
        payload = factory.makeSSOResponse()
        user = UserMgr.from_sso(payload)
        # Does the insert
        user.save(self.db)

        found = UserMgr.find_one(self.db, user.userid)
        self.assertEqual(user.userid, found.userid)
        self.assertEqual(user.fullname, found.fullname)

        found.fullname = 'Changed in test'
        found.save(self.db)

        found2 = UserMgr.find_one(self.db, found.userid)
        self.assertEqual(found.userid, found2.userid)
        self.assertEqual(found.fullname, found2.fullname)

    def test_update_from_sso(self):
        """A new set of SSO data updates the user found."""
        payload = factory.makeSSOResponse()
        user = UserMgr.from_sso(payload)
        # Does the insert
        user.save(self.db)

        new_groups = ['launchpad']
        payload['openid.lp.is_member'] = new_groups[0]
        user.update_from_sso(payload)
        self.assertEqual(new_groups, user.groups)

    def test_auth_principalfinder(self):
        class Request:
            user = UserMgr.from_sso(
                factory.makeSSOResponse(groups=['charmers', 'canonical']))
        expected_principals = [
            'user https://login.ubuntu.com/+id/MSbTCYM',
            'group charmers',
            'group canonical']
        principals = UserMgr.auth_principalfinder('', Request)
        self.assertEqual(expected_principals, principals)


class TestQASet(MongoTestBase):
    """Unit tests for our User object wrapping of the mongo dict."""

    def setUp(self):
        super(TestQASet, self).setUp()
        category = factory.makeQAQuestionGroup()
        self.db.qa.insert(category)

    def test_category_property(self):
        qadata = QAData(self.db)
        self.assertEqual(['reliable'], qadata.categories)

    def test_none_overall_if_no_data(self):
        """When no charm qa data we get None to indicate N/A."""
        qadata = QAData(self.db)
        self.assertIsNone(qadata.score)

    def test_overall_score(self):
        scores = factory.makeQAScore()
        qadata = QAData(self.db, charm_data=scores)

        self.assertEqual(4, qadata.score)

    def test_none_category_score_if_missing_data(self):
        """When no scores are available for the cateogry we get None

        None is our indicator of a score that's N/A

        """
        qadata = QAData(self.db)
        self.assertIsNone(qadata.category_score('reliable'))

    def test_category_score(self):
        scores = factory.makeQAScore()
        qadata = QAData(self.db, charm_data=scores)

        self.assertEqual(4, qadata.category_score('reliable'))

    def test_category_max_score(self):
        """Determine the total available points for the category."""
        qadata = QAData(self.db)
        self.assertEqual(9, qadata.category_max_score('reliable'))


class TestCharmFileSet(MongoTestBase):
    # Escape the base_fileid since all code will escape this. We're comparing
    # against this string so it needs to also be escaped.
    base_fileid = 'charmers/precise/charmworld%2Etest/trunk/'
    charm_data = {
        'series': 'precise',
        'owner': 'charmers',
        'name': 'charmworld.test',
        'bname': 'trunk',
    }
    charm_path = join(
        dirname(dirname(__file__)),
        'testing/data/sample_charm'
    )

    def setUp(self):
        super(TestCharmFileSet, self).setUp()
        self.logger = logging.getLogger(self.__class__.__name__)
        self.log_handler = self.get_handler(self.logger.name)

    def save_files(self, charm_path=None, extra_revisions=[]):
        with charm_branch(charm_path, extra_revisions) as (
                branch_dir, revision):
            charm_data = self.charm_data.copy()
            charm_data['branch_dir'] = branch_dir
            charm_data['store_data'] = {'digest': revision}
            return CharmFileSet.save_files(
                self.fs,
                charm_data,
                branch_dir,
                self.logger)

    def test_fileid_generation(self):
        """A charm must generate an unique fileid."""
        self.assertEqual(
            'charmers/precise/charmworld%2Etest/trunk/readme',
            CharmFileSet.gen_fileid(self.charm_data, 'readme')
        )

    saved_sample_charm_files = [
        'config.yaml',
        'metadata.yaml',
        'README.md',
        'icon.svg',
        'revision',
        'hooks/config-changed',
        'hooks/database-relation-broken',
        'hooks/database-relation-changed',
        'hooks/install',
        'hooks/linkstart',
        'hooks/restart',
        'hooks/start',
        'hooks/stop',
        'hooks/upgrade-charm',
        'hooks/website-relation-changed',
    ]

    def check_bzr_files(self, expected_files, saved_files):
        # There should be the following files stored into the gridfs
        # filestore.
        for name in expected_files:
            fileid = self.base_fileid + quote_key(name)
            self.assertTrue(
                self.fs.exists(fileid),
                'Should find file:' + fileid)
            existing_content = self.fs.get(fileid).read()
            self.assertEqual(expected_files[name], existing_content)
        saved_ids = set(f._id for f in saved_files)
        expected_ids = set(
            self.base_fileid + quote_key(name) for name in expected_files)
        self.assertEqual(expected_ids, saved_ids)

    def expected_files(self):
        result = {}
        sample_data_path = split(split(__file__)[0])[0]
        sample_data_path = join(sample_data_path, 'testing/data/sample_charm')
        for name in self.saved_sample_charm_files:
            result[name] = open(join(sample_data_path, name)).read()
        return result

    def test_load_bzr_files(self):
        """Processing a charm dir loads the file contents."""
        saved = self.save_files()
        self.check_bzr_files(self.expected_files(), saved)

    def test_load_bzr_files_from_old_revision(self):
        """Processing a charm dir loads the file contents."""
        def add_revision_two(cwd, tree):
            path = join(cwd, 'sample_charm/trunk/hooks/new_hook')
            with open(path, 'w') as f:
                f.write('A new hook')
            tree.add(['hooks/new_hook'])
            tree.commit('revision 2')

        saved = self.save_files(extra_revisions=[add_revision_two])
        self.check_bzr_files(self.expected_files(), saved)

    def test_load_bzr_files_from_new_revision(self):
        """Processing a charm dir loads the file contents."""
        def add_revision_two(cwd, tree):
            path = join(cwd, 'sample_charm/trunk/hooks/new_hook')
            with open(path, 'w') as f:
                f.write('A new hook')
            tree.add(['hooks/new_hook'])
            return tree.commit('revision 2')

        saved = self.save_files(extra_revisions=[add_revision_two])
        expected = self.expected_files()
        expected['hooks/new_hook'] = 'A new hook'
        self.check_bzr_files(expected, saved)

    def test_save_files_with_non_existent_bzr_revision(self):
        """Attempts to retrieve a non-existent branch revision result in
        an empty return value of save_files and a log entry."""
        with charm_branch() as (branch_dir, revision):
            charm_data = self.charm_data.copy()
            charm_data['branch_dir'] = branch_dir
            charm_data['store_data'] = {'digest': 'bad-revision-id'}
            result = CharmFileSet.save_files(
                self.fs,
                charm_data,
                branch_dir,
                self.logger)
        self.assertEqual([], result)
        messages = [rec.msg for rec in self.log_handler.buffer]
        self.assertEqual(['Revision bad-revision-id not found'], messages)

    def test_case_kept_for_filenames(self):
        """README.md is lower cased in the key, but upper in filename."""
        self.save_files()

        # Verify that the filename of the README is kept.
        found = self.fs.get_last_version('README.md')
        # fileid is escaped because it's used as a mongo key.
        expected = self.base_fileid + quote_key('README.md')
        self.assertEqual(expected, found._id)

    def test_get_by_id(self):
        """Load a CharmFile instance from the datastore."""
        self.save_files()

        readme_id = self.base_fileid + quote_key('README.md')

        cf = CharmFileSet.get_by_id(self.fs, readme_id)
        self.assertEqual('README.md', cf.filename)
        self.assertEqual(readme_id, cf.fileid)

        # We need to be able to dict the metadata to store in the charm record
        # itself.
        gridout = cf._gridout
        metadata = dict(cf)
        self.assertEqual('README.md', metadata['filename'])
        self.assertEqual(readme_id, metadata['fileid'])
        self.assertEqual(u'charmers/precise/charmworld.test/trunk',
                         metadata['charmid'])
        self.assertEqual(gridout.length, metadata['length'])
        self.assertEqual(gridout.md5, metadata['md5'])
        self.assertEqual('', metadata['subdir'])

    def test_get_by_id_hooks_file(self):
        """The hooks file is a subdir down and we want to track that."""
        self.save_files()

        install_id = self.base_fileid + 'hooks/install'

        cf = CharmFileSet.get_by_id(self.fs, install_id)
        self.assertEqual('install', cf.filename)
        self.assertEqual('hooks', cf.subdir)
        self.assertEqual(install_id, cf.fileid)

        # We need to be able to dict the metadata to store in the charm record
        # itself.
        metadata = dict(cf)
        self.assertEqual('hooks', metadata['subdir'])

    def test_save_files_handles_missing_dir(self):
        path = self.use_context(temp_dir())
        files = self.save_files(path)
        self.assertEqual([], files)

    def test_save_file_ignores_symlink_without_target(self):
        # Symlinks that point to nowhere are not saved.
        path = self.use_context(temp_dir())
        link_path = join(path, 'readme')
        target_name = 'does-not-exist'
        os.symlink(target_name, link_path)
        files = self.save_files(path)
        expected = [
            'File does not exist: readme (resolving to does-not-exist).',
        ]
        messages = [rec.msg for rec in self.log_handler.buffer]
        self.assertEqual(expected, messages)
        self.assertEqual([], files)

    def test_save_file_ignores_symlink_with_outside_target(self):
        # Symlinks that point to a location outside the charm's root
        # directory are also ignored.
        path = self.use_context(temp_dir())
        link_path = join(path, 'readme')
        target_name = '../no-trespassing'
        os.symlink(target_name, link_path)
        files = self.save_files(path)
        expected = [
            "Invalid file path: Path readme points to ../no-trespassing, "
            "which is not inside the charm's root directory"
        ]
        messages = [rec.msg for rec in self.log_handler.buffer]
        self.assertEqual(expected, messages)
        self.assertEqual([], files)

    def test_save_file_ignores_symlink_with_absolute_target_path(self):
        # Symlinks that point to an absolute location are ignored.
        path = self.use_context(temp_dir())
        link_path = join(path, 'readme')
        target_name = '/etc/passwd'
        os.symlink(target_name, link_path)
        files = self.save_files(path)
        expected = [
            "Invalid file path: symlink readme points to the absolute path "
            "/etc/passwd."
        ]
        messages = [rec.msg for rec in self.log_handler.buffer]
        self.assertEqual(expected, messages)
        self.assertEqual([], files)

    def test_save_file_ignores_invalid_icons(self):
        path = self.use_context(temp_dir())
        with file(join(path, 'icon.svg'), 'w') as f:
            f.write('foo')
        files = self.save_files(path)
        self.assertEqual([], files)

    def test_provides_content_type(self):
        path = self.use_context(temp_dir())
        with file(join(path, 'icon.svg'), 'w') as f:
            f.write('<?xml version="1.0"?><svg></svg>')
        with file(join(path, 'readme.txt'), 'w') as f:
            f.write('foo')
        self.save_files(path)
        icon_id = self.base_fileid + quote_key('icon.svg')
        cf = CharmFileSet.get_by_id(self.fs, icon_id)
        self.assertEqual('image/svg+xml', cf.contentType)
        readme_id = self.base_fileid + quote_key('readme.txt')
        cf = CharmFileSet.get_by_id(self.fs, readme_id)
        self.assertEqual('text/plain', cf.contentType)

    def test_directories_match(self):
        path = self.use_context(temp_dir())
        os.mkdir(join(path, 'hooks'))
        os.mkdir(join(path, 'hooks_lib'))
        with file(join(path, 'hooks/a_hook'), 'w') as f:
            f.write('hook')
        with file(join(path, 'hooks_lib/hook_lib'), 'w') as f:
            f.write('not stored in our DB')
        saved = self.save_files(path)
        saved = [charm_file._filename for charm_file in saved]
        self.assertEqual(['a_hook'], saved)


class TestCharmFile(MongoTestBase):

    def test_filename_parsing(self):
        """The filename is the end of the fileid."""
        cf = CharmFile(
            self.fs, '/test/file/path.txt', '/charmers/precise/redis')
        self.assertEqual('path.txt', cf.filename)

    def test_filename_with_caps(self):
        """Whe can specifiy the filename to keep it original."""

        cf = CharmFile(
            self.fs, '/test/file/path.txt', '/charmers/precise/redis',
            filename='PATH.txt')
        self.assertEqual('PATH.txt', cf.filename)

    def test_storing_file_contents(self):
        contents = 'Contents of the file.'
        fileid = '/test/file/path.txt'
        cf = CharmFile(self.fs, fileid, '/charmers/precise/redis')
        cf.save(contents)

        # Make sure we check for the escaped fileid to see it exists.
        self.assertTrue(self.fs.exists(cf.fileid))

        found = self.fs.get(cf.fileid)
        self.assertEqual(contents, found.read())

    def test_save_with_orphan_chunks(self):
        file_id = 'file-id'
        cf = CharmFile(self.fs, file_id, 'baz')
        cf.save('foo')
        # Create orphan chunk
        self.db.fs.files.remove({'_id': file_id})
        self.assertFalse(self.fs.exists(file_id))
        cf.save('bar')
        self.assertEqual('bar', self.fs.get(file_id).read())

    def test_md5(self):
        cf = CharmFile(self.fs, 'file-id', 'baz')
        self.assertIs(None, cf.md5)
        cf.save('asdf')
        self.assertEqual(cf.md5, md5.new('asdf').hexdigest())

    def test_no_contentType(self):
        result = self.fs.put(
            'foo',
            filename='bar',
        )
        instance = self.fs.get(result)
        sentry = object()
        self.assertIs(sentry, getattr(instance, 'contentType', sentry))
        cf = CharmFile(self.fs, None, None, instance=instance)
        self.assertIs(None, cf.contentType)


class TestAcquireSessionSecret(MongoTestBase):

    def test_acquire_session_secret(self):
        """Multiple calls to acquire_session_secret return the same value."""
        secret = acquire_session_secret(self.db)
        self.assertIs(str, type(secret))
        self.assertEqual(secret, acquire_session_secret(self.db))

    def test_db_backed(self):
        """The secret is stored in the database.

        An existing session-secret is used if present.  Deleting it causes a
        new secret to be generated.
        """
        self.db.app_config.insert({'_id': 'session-secret',
                                   'secret': 'arbitrary-string'})
        first_secret = acquire_session_secret(self.db)
        self.assertEqual('arbitrary-string', first_secret)
        self.db.app_config.remove('session-secret')
        second_secret = acquire_session_secret(self.db)
        self.assertNotEqual(first_secret, second_secret)


class TestFindCharms(MongoTestBase):

    def test_find_charms_default(self):
        # By default, find_charms() does not return charms with errors.
        ignore, valid_charm = factory.makeCharm(self.db, name='valid_charm')
        ignore, charm_with_error = factory.makeCharm(
            self.db, name='charm_with_error', charm_error=True)
        result = find_charms(self.db)
        self.assertEqual(1, result.count())
        self.assertEqual(valid_charm, result[0])

    def test_find_charms_invalid_included(self):
        # If the parameter valid_charms_only is set to False, find_charms()
        # returns invalid charms too.
        ignore, valid_charm = factory.makeCharm(self.db, name='valid_charm')
        ignore, charm_with_error = factory.makeCharm(
            self.db, name='charm_with_error', charm_error=True)
        result = find_charms(self.db, valid_charms_only=False)
        self.assertEqual(2, result.count())

    def test_find_charms_with_more_parameters(self):
        # Any parameter of Collection.find() can be specified.
        ignore, charm_1 = factory.makeCharm(self.db, name='valid_charm_1')
        ignore, charm_2 = factory.makeCharm(self.db, name='valid_charm_2')
        result = find_charms(self.db, {'name': 'does-not-exist'})
        self.assertEqual(0, result.count())
        result = find_charms(self.db, spec={'name': 'valid_charm_1'})
        self.assertEqual(1, result.count())
        self.assertEqual(charm_1, result[0])
        result = find_charms(
            self.db, spec={'name': 'valid_charm_1'}, fields=('bname', 'name'))
        expected = {
            '_id': charm_1['_id'],
            'name': charm_1['name'],
            'bname': charm_1['bname']
        }
        self.assertEqual(expected, result[0])

    def test_dont_hide_with_processing_error(self):
        # Charms with errors that occured during the ingest script
        # are returned by find_charms()
        charm_id, charm_data = factory.makeCharm(self.db)
        charm_data['error'] = {}
        self.db.charms.save(charm_data)
        self.assertIn(charm_data,
                      find_charms(self.db, valid_charms_only=False))

    def test_only_promulgated(self):
        ignore, promulgated_charm = factory.makeCharm(
            self.db, promulgated=True)
        factory.makeCharm(self.db)
        result = find_charms(self.db, only_promulgated=True)
        self.assertEqual(1, result.count())
        self.assertEqual(promulgated_charm, result[0])


class TestCharmSource(MongoTestBase):

    def test_sync_index(self):
        self.use_index_client()
        one = factory.makeCharm(self.db)[0]
        two = factory.makeCharm(self.db)[0]
        three = factory.makeCharm(self.db)[0]
        charm_source = CharmSource.from_request(self)
        ids = list(charm_source.sync_index())
        self.assertItemsEqual([one, two, three], ids)
        for charm_id in ids:
            foo, bar = charm_source._get_all(charm_id)
            self.assertEqual(foo, bar)

    def test_sync_index_script(self):
        factory.makeCharm(self.db)[0]
        # Create a charm which is index-incompatible with the previous one.
        factory.makeCharm(self.db,
                          owner={'first': 'J', 'last': 'Random'})[0]
        handler = self.get_handler('sync-index')
        with patch('charmworld.models.configure_logging'):
            with patch('charmworld.models.get_ini', lambda: INI):
                with force_index_client(self.use_index_client()):
                    sync_index()
            messages = [r.getMessage() for r in handler.buffer]
            self.assertIn('Unable to index charm.', messages)
        self.assertEqual(1, len(self.index_client.api_search()))

    def test_save_deletes_on_error(self):
        self.use_index_client()
        source = CharmSource.from_request(self)
        source.save({'_id': 'a', 'b': {}})
        with self.assertRaises(ElasticHttpError):
            source.save({'_id': 'a', 'b': 'c'})
        with self.assertRaises(ElasticHttpNotFoundError):
            self.index_client.get('a')


class TestOptionsStorage(TestCase):

    yaml = {
        'foo': {
            'default': 'bar',
            'type': 'string',
            'description': 'Hello.'
        },
        'baz': {
            'default': 42,
            'type': 'int',
            'description': 'Hello.'
        },
    }
    storage = [{
        'name': 'baz',
        'default': 42,
        'type': 'int',
        'description': 'Hello.'
    }, {
        'name': 'foo',
        'default': 'bar',
        'type': 'string',
        'description': 'Hello.'
    }]

    def test_options_to_storage(self):
        self.assertItemsEqual(self.storage, options_to_storage(self.yaml))

    def test_options_to_storage_none(self):
        self.assertItemsEqual([], options_to_storage(None))

    def test_storage_to_options(self):
        self.assertEqual(self.yaml, storage_to_options(self.storage))

    def test_storage_to_options_duplicate_key(self):
        with self.assertRaises(ValueError) as exc_context:
            storage_to_options([{'name': 'same'}, {'name': 'same'}])
        self.assertEqual('Option name "same" occurs multiple times.',
                         str(exc_context.exception))

    def test_roundtrip(self):
        self.assertEqual(
            self.yaml, storage_to_options(options_to_storage(self.yaml)))
        self.assertItemsEqual(
            self.storage,
            options_to_storage(storage_to_options(self.storage)))


class BundleTestCase(TestCase):
    """Unit tests for the Bundle model that wraps bundle representations."""

    def test_init_with_empty_representation(self):
        bundle_data = {}
        bundle = Bundle(bundle_data)
        self.assertIs(bundle_data, bundle._raw_representation)
        self.assertEqual(bundle._COMMON_REPRESENTATION, bundle._representation)

    def test_init_with_minimal_data(self):
        bundle_data = {
            'owner': 'sinzui',
            'basket': 'mysql',
            'name': 'tiny',
            'inherits': 'main-mysql',
            'series': 'series',
            'title': 'Tiny bundle',
            'description': 'My Tiny Bundle',
            'services': dict(apache=1),
            'relations': dict(a=1),
            'promulgated': True,
            'branch_deleted': True,
        }
        bundle = Bundle(bundle_data)
        self.assertIs(bundle_data, bundle._raw_representation)
        self.assertEqual(bundle_data, bundle._representation)

    def test_reject_arbitrary_attributes(self):
        bundle = Bundle({})
        with self.assertRaises(AttributeError):
            bundle.not_a_real_attribute

    def test_is_equal(self):
        bundle1 = Bundle(
            {'name': 'tiny',
             'owner': 'bac',
             'basket': 'apache-1',
             'inherits': 'big-apache',
             })
        bundle2 = Bundle(
            {'name': 'tiny',
             'owner': 'bac',
             'basket': 'apache-1',
             'inherits': 'moderately-big-apache',
             'branch_deleted': True,
             })
        self.assertEqual(bundle1, bundle2)

    def test_not_equal(self):
        bundle1 = Bundle(
            {'name': 'tiny',
             'owner': 'bac',
             'basket': 'apache-4',
             })
        bundle2 = Bundle(
            {'name': 'tiny',
             'owner': 'abc',
             'basket': 'apache-4',
             })
        self.assertNotEqual(bundle1, bundle2)

    def test_id(self):
        bundle = Bundle(
            {'name': 'tiny',
             'owner': 'bac',
             'basket': 'apache-8',
             })
        self.assertEqual(
            "~bac/apache-8/tiny", bundle.id)

    def test_perm_url(self):
        bundle = Bundle(
            {'name': 'tiny',
             'owner': 'bac',
             'basket': 'apache-6',
             })
        self.assertEqual(
            "jc:~bac/apache-6/tiny", bundle.permanent_url)