def test_removed(self):
     # At least one public app must exist for dump_apps to run.
     app_factory(name='second app', status=mkt.STATUS_PUBLIC)
     app_path = os.path.join(self.export_directory, self.app_path)
     app = Webapp.objects.get(pk=337141)
     app.update(status=mkt.STATUS_PUBLIC)
     self.create_export('tarball-name')
     assert private_storage.exists(app_path)
     app.update(status=mkt.STATUS_PENDING)
     self.create_export('tarball-name')
     assert not private_storage.exists(app_path)

 def test_delete_with_file(self):
     """Test that when a Extension instance is deleted, the corresponding
     file on the filesystem is also deleted."""
     extension = Extension.objects.create(version='0.1')
     file_path = extension.file_path
     with private_storage.open(file_path, 'w') as f:
         f.write('sample data\n')
     assert private_storage.exists(file_path)
     try:
         extension.delete()
         assert not private_storage.exists(file_path)
     finally:
         if private_storage.exists(file_path):
             private_storage.delete(file_path)

 def reviewer_sign_file(self):
     """Sign the original file (`file_path`) with reviewer certs, then move
     the signed file to the reviewers-specific signed path
     (`reviewer_signed_file_path`) on private storage."""
     if not self.extension.uuid:
         raise SigningError('Need uuid to be set to sign')
     if not self.pk:
         raise SigningError('Need version pk to be set to sign')
     ids = json.dumps({
         'id': self.review_id,
         'version': self.pk
     })
     with statsd.timer('extensions.sign_reviewer'):
         try:
             # This will read the file from self.file_path, generate a
             # reviewer signature and write the signed file to
             # self.reviewer_signed_file_path.
             sign_app(private_storage.open(self.file_path),
                      self.reviewer_signed_file_path, ids, reviewer=True)
         except SigningError:
             log.info(
                 '[ExtensionVersion:%s] Reviewer Signing failed' % self.pk)
             if private_storage.exists(self.reviewer_signed_file_path):
                 private_storage.delete(self.reviewer_signed_file_path)
             raise

    def test_upload_sign_error_existing(self, sign_app_mock):
        sign_app_mock.side_effect = SigningError
        langpack = self.create_langpack()
        eq_(LangPack.objects.count(), 1)
        original_uuid = langpack.uuid
        original_file_path = langpack.file_path
        original_file_version = langpack.file_version
        original_version = langpack.version
        # create_langpack() doesn't create a fake file, let's add one.
        with public_storage.open(langpack.file_path, 'w') as f:
            f.write('.')
        upload = self.upload('langpack')
        with self.assertRaises(SigningError):
            LangPack.from_upload(upload, instance=langpack)
        # Test that we didn't delete the upload file
        ok_(private_storage.exists(upload.path))
        # Test that we didn't delete the existing filename or alter the
        # existing langpack in the database.
        eq_(LangPack.objects.count(), 1)
        langpack.reload()
        eq_(original_uuid, langpack.uuid)
        eq_(langpack.file_path, original_file_path)
        eq_(original_file_version, langpack.file_version)
        eq_(original_version, langpack.version)
        ok_(public_storage.exists(langpack.file_path))

        # Cleanup
        public_storage.delete(langpack.file_path)

 def reviewer_sign_file(self):
     """Sign the original file (`file_path`) with reviewer certs, then move
     the signed file to the reviewers-specific signed path
     (`reviewer_signed_file_path`) on private storage."""
     if not self.extension.uuid:
         raise SigningError('Need uuid to be set to sign')
     if not self.pk:
         raise SigningError('Need version pk to be set to sign')
     ids = json.dumps({
         # Reviewers get a unique 'id' so the reviewer installed add-on
         # won't conflict with the public add-on, and also so even multiple
         # versions of the same add-on can be installed side by side with
         # other versions.
         'id': 'reviewer-{guid}-{version_id}'.format(
             guid=self.extension.uuid, version_id=self.pk),
         'version': self.pk
     })
     with statsd.timer('extensions.sign_reviewer'):
         try:
             # This will read the file from self.file_path, generate a
             # reviewer signature and write the signed file to
             # self.reviewer_signed_file_path.
             sign_app(private_storage.open(self.file_path),
                      self.reviewer_signed_file_path, ids, reviewer=True)
         except SigningError:
             log.info(
                 '[ExtensionVersion:%s] Reviewer Signing failed' % self.pk)
             if private_storage.exists(self.reviewer_signed_file_path):
                 private_storage.delete(self.reviewer_signed_file_path)
             raise

 def test_delete_no_file(self):
     """Test that the Extension instance can be deleted without the file
     being present."""
     extension = Extension.objects.create(version='0.1')
     filename = extension.file_path
     assert not private_storage.exists(filename)
     extension.delete()

 def test_upload_sign_error(self, sign_app_mock):
     sign_app_mock.side_effect = SigningError
     eq_(LangPack.objects.count(), 0)
     upload = self.upload('langpack')
     with self.assertRaises(SigningError):
         LangPack.from_upload(upload)
     # Test that we didn't delete the upload file
     ok_(private_storage.exists(upload.path))

 def test_move(self):
     src = self.newfile('src.txt', '<contents>')
     dst = self.path('somedir/dst.txt')
     move_stored_file(
         src, dst,
         src_storage=private_storage, dst_storage=private_storage)
     eq_(self.contents(dst), '<contents>')
     eq_(private_storage.exists(src), False)

 def test_delete_with_file(self):
     """Test that when a Extension instance is deleted, the ExtensionVersion
     referencing it are also deleted, as well as the attached files."""
     extension = Extension.objects.create()
     version = ExtensionVersion.objects.create(
         extension=extension, version='0.1')
     file_path = version.file_path
     with private_storage.open(file_path, 'w') as f:
         f.write('sample data\n')
     assert private_storage.exists(file_path)
     try:
         extension.delete()
         assert not Extension.objects.count()
         assert not ExtensionVersion.objects.count()
         assert not private_storage.exists(file_path)
     finally:
         if private_storage.exists(file_path):
             private_storage.delete(file_path)

    def test_promo_img_too_small(self):
        with local_storage.open(get_image_path('preview.jpg')) as f:
            errors, upload_hash = check_upload(f, 'promo_img', 'image/png')
            ok_(errors)
            ok_(upload_hash)

            tmp_img_path = os.path.join(settings.TMP_PATH, 'promo_img',
                                        upload_hash)
            ok_(private_storage.exists(tmp_img_path))

    def test_icon_ok(self):
        with local_storage.open(get_image_path('mozilla-sq.png')) as f:
            errors, upload_hash = check_upload(f, 'icon', 'image/png')
            ok_(not errors)
            ok_(upload_hash)

            tmp_img_path = os.path.join(settings.TMP_PATH, 'icon',
                                        upload_hash)
            ok_(private_storage.exists(tmp_img_path))

    def setup_files(self):
        # Clean out any left over stuff.
        private_storage.delete(self.file.signed_file_path)
        private_storage.delete(self.file.signed_reviewer_file_path)

        # Make sure the source file is there.
        if not private_storage.exists(self.file.file_path):
            copy_to_storage(self.packaged_app_path('mozball.zip'),
                            self.file.file_path)

def move_files_to_their_new_locations(apps, schema_editor):
    ExtensionVersion = apps.get_model('extensions', 'ExtensionVersion')
    versions = ExtensionVersion.objects.all()
    for version in versions:
        # We lost the version number on old deleted versions, nothing we
        # can do about those. It's fine.
        if version.deleted:
            continue

        # Migrations have no access to custom properties and methods, so we
        # have to re-generate file paths.
        unsigned_prefix = os.path.join(
            settings.EXTENSIONS_PATH, str(version.extension.pk))
        signed_prefix = os.path.join(
            settings.SIGNED_EXTENSIONS_PATH, str(version.extension.pk))
        signed_reviewer_prefix = os.path.join(
            settings.EXTENSIONS_PATH, str(version.extension.pk), 'reviewers')
        filename = 'extension-%s.zip' % version.version

        # Original paths have the version number in them.
        original_unsigned_file_path = os.path.join(unsigned_prefix, filename)
        original_signed_file_path = os.path.join(signed_prefix, filename)
        original_reviewer_signed_file_path = os.path.join(
            signed_reviewer_prefix, filename)

        # New paths use the version pk instead, which will always be available.
        new_filename = 'extension-%s.zip' % version.pk
        new_unsigned_file_path = os.path.join(unsigned_prefix, new_filename)
        new_signed_file_path = os.path.join(signed_prefix, new_filename)
        new_reviewer_signed_file_path = os.path.join(
            signed_reviewer_prefix, new_filename)

        # Do the actual moving.
        if private_storage.exists(original_unsigned_file_path):
            move_stored_file(
                original_unsigned_file_path, new_unsigned_file_path)
        if private_storage.exists(original_reviewer_signed_file_path):
            move_stored_file(
                original_reviewer_signed_file_path,
                new_reviewer_signed_file_path)
        if public_storage.exists(original_signed_file_path):
            move_stored_file(
                original_signed_file_path, new_signed_file_path,
                src_storage=public_storage, dst_storage=public_storage)

 def test_upload_new(self):
     eq_(Extension.objects.count(), 0)
     upload = self.upload('extension')
     extension = Extension.from_upload(upload)
     eq_(extension.version, '0.1')
     eq_(extension.name, u'My Lîttle Extension')
     eq_(extension.default_language, 'en-GB')
     eq_(extension.slug, u'my-lîttle-extension')
     eq_(extension.filename, 'extension-%s.zip' % extension.version)
     ok_(extension.filename in extension.file_path)
     ok_(extension.file_path.startswith(extension.path_prefix))
     ok_(private_storage.exists(extension.file_path))
     eq_(extension.manifest, self.expected_manifest)
     eq_(Extension.objects.count(), 1)

 def test_upload_new(self):
     eq_(Extension.objects.count(), 0)
     upload = self.upload('extension')
     extension = Extension.from_upload(upload, user=self.user)
     eq_(extension.version, '0.1')
     eq_(list(extension.authors.all()), [self.user])
     eq_(extension.name, u'My Lîttle Extension')
     eq_(extension.default_language, 'en-GB')
     eq_(extension.slug, u'my-lîttle-extension')
     eq_(extension.filename, 'extension-%s.zip' % extension.version)
     ok_(extension.filename in extension.file_path)
     ok_(private_storage.exists(extension.file_path))
     eq_(extension.manifest, self.expected_manifest)
     eq_(Extension.objects.count(), 1)

    def handle_file_operations(self, upload):
        """Copy the file attached to a FileUpload to the Extension instance."""
        upload.path = smart_path(nfd_str(upload.path))

        if private_storage.exists(self.file_path):
            # The filename should not exist. If it does, it means we are trying
            # to re-upload the same version. This should have been caught
            # before, so just raise an exception.
            raise RuntimeError(
                'Trying to upload a file to a destination that already exists')

        # Copy file from fileupload. This uses private_storage for now as the
        # unreviewed, unsigned filename is private.
        copy_stored_file(
            upload.path, self.file_path,
            src_storage=private_storage, dst_storage=private_storage)

 def test_file_order(self):
     self.viewer.extract()
     dest = self.viewer.dest
     private_storage.open(os.path.join(dest, 'manifest.webapp'),
                          'w').close()
     subdir = os.path.join(dest, 'chrome')
     with private_storage.open(os.path.join(subdir, 'foo'), 'w') as f:
         f.write('.')
     if not private_storage.exists(subdir):
         # Might be on S3, which doesn't have directories (and
         # django-storages doesn't support empty files).
         with private_storage.open(subdir, 'w') as f:
             f.write('.')
     cache.clear()
     files = self.viewer.get_files().keys()
     rt = files.index(u'chrome')
     eq_(files[rt:rt + 3], [u'chrome', u'chrome/foo', u'dictionaries'])

    def upload(self, name, **kwargs):
        if os.path.splitext(name)[-1] not in ['.webapp', '.zip']:
            name = name + '.zip'

        v = json.dumps(dict(errors=0, warnings=1, notices=2, metadata={}))
        fname = nfd_str(self.packaged_app_path(name))
        if not local_storage.exists(fname):
            raise ValueError('The file %s does not exist :(', fname)
        if not private_storage.exists(fname):
            copy_to_storage(fname)
        data = {
            'path': fname,
            'name': name,
            'hash': 'sha256:%s' % name,
            'validation': v
        }
        data.update(**kwargs)
        return FileUpload.objects.create(**data)

    def _test_create_success(self, client):
        headers = {
            'HTTP_CONTENT_TYPE': 'application/zip',
            'HTTP_CONTENT_DISPOSITION': 'form-data; name="binary_data"; '
                                        'filename="foo.zip"'
        }
        with open(self.packaged_app_path('extension.zip'), 'rb') as fd:
            response = client.post(self.list_url, fd.read(),
                                   content_type='application/zip', **headers)

        eq_(response.status_code, 202)
        data = response.json
        upload = FileUpload.objects.get(pk=data['id'])
        eq_(upload.valid, True)  # We directly set uploads as valid atm.
        eq_(upload.name, 'foo.zip')
        ok_(upload.hash.startswith('sha256:58ef3f15dd423c3ab9b0285ac01e692c5'))
        ok_(upload.path)
        ok_(private_storage.exists(upload.path))
        return upload

    def _is_binary(self, mimetype, path):
        """Uses the filename to see if the file can be shown in HTML or not."""
        # Re-use the blocked data from amo-validator to spot binaries.
        ext = os.path.splitext(path)[1][1:]
        if ext in blocked_extensions:
            return True

        # S3 will return false for storage.exists() for directory paths, so
        # os.path call is safe here.
        if private_storage.exists(path) and not os.path.isdir(path):
            with private_storage.open(path, 'r') as rfile:
                bytes = tuple(map(ord, rfile.read(4)))
            if any(bytes[:len(x)] == x for x in blocked_magic_numbers):
                return True

        if mimetype:
            major, minor = mimetype.split('/')
            if major == 'image':
                return 'image'  # Mark that the file is binary, but an image.

        return False