def from_upload(cls, upload, version, parse_data={}):
        upload.path = smart_path(nfd_str(upload.path))
        ext = os.path.splitext(upload.path)[1]

        f = cls(version=version)
        f.filename = f.generate_filename(extension=ext or '.zip')
        f.size = private_storage.size(upload.path)  # Size in bytes.
        f.status = mkt.STATUS_PENDING
        # Re-use the file-upload hash if we can, no need to regenerate a new
        # one if we can avoid that.
        f.hash = upload.hash or f.generate_hash(upload.path)
        f.save()

        log.debug('New file: %r from %r' % (f, upload))

        # Move the uploaded file from the temp location.
        copy_stored_file(
            upload.path,
            os.path.join(version.path_prefix, nfd_str(f.filename)),
            src_storage=private_storage,
            dst_storage=private_storage)

        if upload.validation:
            FileValidation.from_json(f, upload.validation)

        return f

    def test_pngcrush_image_is_called(self, mock_move):
        expected_suffix = '.opti.png'
        tmp_src = tempfile.NamedTemporaryFile(suffix='.png', delete=False)
        tmp_dest = os.path.splitext(tmp_src.name)[0] + expected_suffix
        copy_stored_file(self.img_path, tmp_dest, src_storage=public_storage,
                         dst_storage=local_storage)
        with mock.patch('tempfile.NamedTemporaryFile',
                        lambda *a, **k: tmp_src):
            rval = tasks.pngcrush_image(self.img_path)
            # pngcrush_image copies the stored file to a local tempfile.
            eq_(open(tmp_src.name).read(),
                public_storage.open(self.img_path).read())

        expected_cmd = ['pngcrush', '-q', '-rem', 'alla', '-brute', '-reduce',
                        '-e', expected_suffix, tmp_src.name]
        # pngcrush_image incokes pngcrush with the tempfile name and writes to
        # another tempfile.
        self.mock_popen.assert_called_once_with(
            expected_cmd, stdin=subprocess.PIPE, stderr=subprocess.PIPE,
            stdout=subprocess.PIPE)
        # The output file is then copied back to storage.
        mock_move.assert_called_once_with(tmp_dest, self.img_path,
                                          dst_storage=public_storage,
                                          src_storage=local_storage)
        eq_(rval, {'image_hash': 'bb362450'})

 def test_self_copy(self):
     src = self.newfile('src.txt', '<contents>')
     dst = self.path('src.txt')
     copy_stored_file(
         src, dst,
         src_storage=private_storage, dst_storage=private_storage)
     eq_(self.contents(dst), '<contents>')

    def setUp(self):
        self.app = Webapp.objects.get(pk=337141)
        self.app.update(is_packaged=True, status=mkt.WEBAPPS_UNREVIEWED_STATUS)
        self.dev = self.app.authors.all()[0]
        self.regular = UserProfile.objects.get(pk=999)
        self.version = self.app.versions.latest()
        self.file = self.version.all_files[0]

        self.versions = [self.version,
                         self.app.versions.create(
                             version='%s.1' % self.version.version)]

        self.files = [self.file,
                      File.objects.create(version=self.versions[1],
                                          filename='webapp.zip')]

        self.login_as_editor()

        for file_obj in self.files:
            src = os.path.join(settings.ROOT, packaged_app)
            if file_obj.status in mkt.LISTED_STATUSES:
                target = public_storage
            else:
                target = private_storage
            copy_stored_file(src, file_obj.file_path,
                             src_storage=local_storage,
                             dst_storage=target)

        self.file_viewer = FileViewer(self.file)

 def setUp(self):
     fn = get_file('dictionary-test.xpi')
     if storage_is_remote():
         copy_stored_file(
             fn, fn,
             src_storage=local_storage, dst_storage=private_storage)
     self.viewer = FileViewer(make_file(1, fn))

def run_validator(file_path, url=None):
    """A pre-configured wrapper around the app validator."""

    temp_path = None
    # Make a copy of the file since we can't assume the
    # uploaded file is on the local filesystem.
    temp_path = tempfile.mktemp()
    copy_stored_file(
        file_path, temp_path,
        src_storage=private_storage, dst_storage=local_storage)

    with statsd.timer('mkt.developers.validator'):
        is_packaged = zipfile.is_zipfile(temp_path)
        if is_packaged:
            log.info(u'Running `validate_packaged_app` for path: %s'
                     % (file_path))
            with statsd.timer('mkt.developers.validate_packaged_app'):
                return validate_packaged_app(
                    temp_path,
                    market_urls=settings.VALIDATOR_IAF_URLS,
                    timeout=settings.VALIDATOR_TIMEOUT,
                    spidermonkey=settings.SPIDERMONKEY)
        else:
            log.info(u'Running `validate_app` for path: %s' % (file_path))
            with statsd.timer('mkt.developers.validate_app'):
                return validate_app(open(temp_path).read(),
                                    market_urls=settings.VALIDATOR_IAF_URLS,
                                    url=url)

    # Clean up copied files.
    os.unlink(temp_path)

 def test_non_ascii(self):
     src = self.newfile(u'kristi\u0107.txt',
                        u'ivan kristi\u0107'.encode('utf8'))
     dst = self.path(u'somedir/kristi\u0107.txt')
     copy_stored_file(
         src, dst,
         src_storage=private_storage, dst_storage=private_storage)
     eq_(self.contents(dst), 'ivan kristi\xc4\x87')

 def setUp(self):
     self.webapp_path = tempfile.mktemp(suffix='.webapp')
     copy_stored_file(
         self.manifest_path('mozball.webapp'), self.webapp_path,
         src_storage=local_storage, dst_storage=private_storage)
     self.tmp_files = []
     self.manifest = dict(name=u'Ivan Krsti\u0107', version=u'1.0',
                          description=u'summary',
                          developer=dict(name=u'Dev Namé'))

 def test_preview_size(self):
     name = 'non-animated.gif'
     form = forms.PreviewForm({'upload_hash': name, 'position': 1})
     copy_stored_file(
         get_image_path(name), os.path.join(self.dest, name),
         src_storage=local_storage, dst_storage=private_storage)
     assert form.is_valid(), form.errors
     form.save(self.addon)
     eq_(self.addon.previews.all()[0].sizes,
         {u'image': [250, 297], u'thumbnail': [100, 119]})

    def setup_files(self):
        # Clean out any left over stuff.
        public_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_stored_file(self.packaged_app_path('mozball.zip'),
                             self.file.file_path, src_storage=local_storage,
                             dst_storage=private_storage)

 def test_disabled_but_owner(self):
     self.login('[email protected]')
     self.file.update(status=mkt.STATUS_DISABLED)
     copy_stored_file(self.packaged_app_path('mozball.zip'),
                      self.file.file_path,
                      src_storage=local_storage,
                      dst_storage=private_storage)
     path = private_storage.url(self.file.file_path)
     res = self.client.get(self.url)
     self.assert3xx(res, path)

 def test_no_manifest_at_root(self):
     path = self.packaged_app_path('no-manifest-at-root.zip')
     if storage_is_remote():
         copy_stored_file(path, path, src_storage=local_storage,
                          dst_storage=private_storage)
     with self.assertRaises(forms.ValidationError) as exc:
         WebAppParser().parse(private_storage.open(path))
     m = exc.exception.messages[0]
     assert m.startswith('The file "manifest.webapp" was not found'), (
         'Unexpected: %s' % m)

    def get_image(self, filename):
        """Copy image to tmp and return tmp path.

        We do this because the task `resize_preview` removes the src file when
        finished.

        """
        src = get_image_path(filename)
        dst = os.path.join(settings.TMP_PATH, 'preview', filename)
        copy_stored_file(
            src, dst, src_storage=local_storage, dst_storage=private_storage)
        return dst

def resize_video(src, pk, user_pk=None, **kw):
    """Try and resize a video and cope if it fails."""
    instance = Preview.objects.get(pk=pk)
    user = UserProfile.objects.get(pk=user_pk) if user_pk else None
    try:
        copy_stored_file(src, src, src_storage=private_storage,
                         dst_storage=local_storage)
        result = _resize_video(src, instance, **kw)
    except Exception, err:
        log.error('Error on processing video: %s' % err)
        _resize_error(src, instance, user)
        raise

 def test_disabled_but_admin(self):
     self.login("[email protected]")
     self.file.update(status=mkt.STATUS_DISABLED)
     copy_stored_file(
         self.packaged_app_path("mozball.zip"),
         self.file.file_path,
         src_storage=local_storage,
         dst_storage=private_storage,
     )
     path = private_storage.url(self.file.file_path)
     res = self.client.get(self.url)
     self.assert3xx(res, path)

def compress_export(tarball_name, date):
    # We need a temporary directory on the local filesystem that will contain
    # all files in order to call `tar`.
    local_source_dir = tempfile.mkdtemp()

    apps_dirpath = os.path.join(settings.DUMPED_APPS_PATH, 'apps')

    # In case apps_dirpath is empty, add a dummy file to make the apps
    # directory in the tar archive non-empty. It should not happen in prod, but
    # it's nice to have it to prevent the task from failing entirely.
    with private_storage.open(
            os.path.join(apps_dirpath, '0', '.keep'), 'w') as fd:
        fd.write('.')

    # Now, copy content from private_storage to that temp directory. We don't
    # need to worry about creating the directories locally, the storage class
    # does that for us.
    for dirpath, dirnames, filenames in walk_storage(
            apps_dirpath, storage=private_storage):
        for filename in filenames:
            src_path = os.path.join(dirpath, filename)
            dst_path = os.path.join(
                local_source_dir, 'apps', os.path.basename(dirpath), filename)
            copy_stored_file(
                src_path, dst_path, src_storage=private_storage,
                dst_storage=local_storage)

    # Also add extra files to the temp directory.
    extra_filenames = compile_extra_files(local_source_dir, date)

    # All our files are now present locally, let's generate a local filename
    # that will contain the final '.tar.gz' before it's copied over to
    # public storage.
    local_target_file = tempfile.NamedTemporaryFile(
        suffix='.tgz', prefix='dumped-apps-')

    # tar ALL the things!
    cmd = ['tar', 'czf', local_target_file.name, '-C',
           local_source_dir] + ['apps'] + extra_filenames
    task_log.info(u'Creating dump {0}'.format(local_target_file.name))
    subprocess.call(cmd)

    # Now copy the local tgz to the public storage.
    remote_target_filename = os.path.join(
        settings.DUMPED_APPS_PATH, 'tarballs', '%s.tgz' % tarball_name)
    copy_stored_file(local_target_file.name, remote_target_filename,
                     src_storage=local_storage,
                     dst_storage=public_storage)

    # Clean-up.
    local_target_file.close()
    rm_directory(local_source_dir)
    return remote_target_filename

 def test_parse_packaged_BOM(self):
     path = self.packaged_app_path('mozBOM.zip')
     if storage_is_remote():
         copy_stored_file(path, path, src_storage=local_storage,
                          dst_storage=private_storage)
     wp = WebAppParser().parse(private_storage.open(path))
     eq_(wp['guid'], None)
     eq_(wp['name']['en-US'], u'Packaged MozBOM ょ')
     eq_(wp['description']['en-US'], u'Exciting BOM action!')
     eq_(wp['description']['es'], u'¡Acción BOM!')
     eq_(wp['description']['it'], u'Azione BOM!')
     eq_(wp['version'], '1.0')
     eq_(wp['default_locale'], 'en-US')

 def test_preview_size(self):
     name = 'non-animated.gif'
     form = forms.PreviewForm({'upload_hash': name, 'position': 1})
     copy_stored_file(
         get_image_path(name), os.path.join(self.dest, name),
         src_storage=local_storage, dst_storage=private_storage)
     assert form.is_valid(), form.errors
     form.save(self.addon)
     # Since the task is a post-request-task and we are outside the normal
     # request-response cycle, manually send the tasks.
     post_request_task._send_tasks()
     eq_(self.addon.previews.all()[0].sizes,
         {u'image': [250, 297], u'thumbnail': [100, 119]})

 def setUp(self):
     self.img_path = tempfile.mktemp()
     copy_stored_file(get_image_path('mozilla.png'),
                      self.img_path,
                      src_storage=local_storage,
                      dest_storage=public_storage)
     patcher = mock.patch('subprocess.Popen')
     self.mock_popen = patcher.start()
     attrs = {
         'returncode': 0,
         'communicate.return_value': ('ouput', 'error')
     }
     self.mock_popen.return_value.configure_mock(**attrs)
     self.addCleanup(patcher.stop)

 def setUp(self):
     super(TestTask, self).setUp()
     self.app = Webapp.objects.get(pk=337141)
     self.preview = Preview.objects.create(
         addon=self.app, thumbnail_path=tempfile.mkstemp()[1],
         image_path=tempfile.mkstemp()[1])
     # Copy files to private storage where `resize_video` expects it.
     self.tmp_good = tempfile.NamedTemporaryFile(suffix='.webm').name
     self.tmp_bad = tempfile.NamedTemporaryFile(suffix='.png').name
     copy_stored_file(files['good'], self.tmp_good,
                      src_storage=local_storage,
                      dst_storage=private_storage)
     copy_stored_file(files['bad'], self.tmp_bad,
                      src_storage=local_storage,
                      dst_storage=private_storage)