src/chrome/common/extensions/docs/server2/new_github_file_system_test.py

   1 #!/usr/bin/env python
   2 # Copyright 2013 The Chromium Authors. All rights reserved.
   3 # Use of this source code is governed by a BSD-style license that can be
   4 # found in the LICENSE file.
   5
   6 import json
   7 from cStringIO import StringIO
   8 import unittest
   9 from zipfile import ZipFile
  10
  11 from caching_file_system import CachingFileSystem
  12 from file_system import FileNotFoundError, StatInfo
  13 from fake_url_fetcher import FakeURLFSFetcher
  14 from local_file_system import LocalFileSystem
  15 from new_github_file_system import GithubFileSystem
  16 from object_store_creator import ObjectStoreCreator
  17 from test_file_system import TestFileSystem
  18
  19 # A made up sha hash value.
  20 FAKE_HASH = '7becb9f554dec76bd0fc12c1d32dbaff1d134a4d'
  21
  22
  23 def _ZipFromFiles(file_dict):
  24   string = StringIO()
  25   zipfile = ZipFile(string, 'w')
  26   for filename, contents in file_dict.iteritems():
  27     zipfile.writestr(filename, contents)
  28   zipfile.close()
  29
  30   return string.getvalue()
  31
  32
  33 class TestGithubFileSystem(unittest.TestCase):
  34   def setUp(self):
  35     self._gfs = GithubFileSystem.ForTest('repo', FakeURLFSFetcher.CreateLocal())
  36     # Start and finish the repository load.
  37     self._cgfs = CachingFileSystem(
  38         self._gfs, ObjectStoreCreator.ForTest())
  39
  40   def testReadDirectory(self):
  41     self._gfs.Refresh().Get()
  42     self.assertEqual(
  43         sorted(['requirements.txt', '.gitignore', 'README.md', 'src/']),
  44         sorted(self._gfs.ReadSingle('').Get()))
  45     self.assertEqual(
  46         sorted(['__init__.notpy', 'hello.notpy']),
  47         sorted(self._gfs.ReadSingle('src/').Get()))
  48
  49   def testReadFile(self):
  50     self._gfs.Refresh().Get()
  51     expected = (
  52       '# Compiled Python files\n'
  53       '*.pyc\n'
  54     )
  55     self.assertEqual(expected, self._gfs.ReadSingle('.gitignore').Get())
  56
  57   def testMultipleReads(self):
  58     self._gfs.Refresh().Get()
  59     self.assertEqual(
  60         self._gfs.ReadSingle('requirements.txt').Get(),
  61         self._gfs.ReadSingle('requirements.txt').Get())
  62
  63   def testReads(self):
  64     self._gfs.Refresh().Get()
  65     expected = {
  66         'src/': sorted(['hello.notpy', '__init__.notpy']),
  67         '': sorted(['requirements.txt', '.gitignore', 'README.md', 'src/'])
  68     }
  69
  70     read = self._gfs.Read(['', 'src/']).Get()
  71     self.assertEqual(expected['src/'], sorted(read['src/']))
  72     self.assertEqual(expected[''], sorted(read['']))
  73
  74   def testStat(self):
  75     self._gfs.Refresh().Get()
  76     dir_stat = StatInfo(FAKE_HASH, {
  77       'hello.notpy': StatInfo(FAKE_HASH),
  78       '__init__.notpy': StatInfo(FAKE_HASH)
  79     })
  80
  81     self.assertEqual(StatInfo(FAKE_HASH), self._gfs.Stat('README.md'))
  82     self.assertEqual(StatInfo(FAKE_HASH), self._gfs.Stat('src/hello.notpy'))
  83     self.assertEqual(dir_stat, self._gfs.Stat('src/'))
  84
  85   def testBadReads(self):
  86     self._gfs.Refresh().Get()
  87     self.assertRaises(FileNotFoundError, self._gfs.Stat, 'DONT_README.md')
  88     self.assertRaises(FileNotFoundError,
  89                       self._gfs.ReadSingle('DONT_README.md').Get)
  90
  91   def testCachingFileSystem(self):
  92     self._cgfs.Refresh().Get()
  93     initial_cgfs_read_one = self._cgfs.ReadSingle('src/hello.notpy').Get()
  94
  95     self.assertEqual(initial_cgfs_read_one,
  96                      self._gfs.ReadSingle('src/hello.notpy').Get())
  97     self.assertEqual(initial_cgfs_read_one,
  98                      self._cgfs.ReadSingle('src/hello.notpy').Get())
  99
 100     initial_cgfs_read_two = self._cgfs.Read(
 101         ['README.md', 'requirements.txt']).Get()
 102
 103     self.assertEqual(
 104         initial_cgfs_read_two,
 105         self._gfs.Read(['README.md', 'requirements.txt']).Get())
 106     self.assertEqual(
 107         initial_cgfs_read_two,
 108         self._cgfs.Read(['README.md', 'requirements.txt']).Get())
 109
 110   def testWithoutRefresh(self):
 111     self.assertRaises(FileNotFoundError, self._gfs.ReadSingle('src/').Get)
 112
 113   def testRefresh(self):
 114     def make_sha_json(hash_value):
 115       from copy import deepcopy
 116       commit_json = json.loads(deepcopy(LocalFileSystem('').ReadSingle(
 117           'test_data/github_file_system/test_owner/repo/commits/HEAD').Get()))
 118       commit_json['commit']['tree']['sha'] = 'hash' + FAKE_HASH[4:]
 119
 120       return json.dumps(commit_json)
 121
 122     files = {
 123       'zipfile/': '',
 124       'zipfile/hello.txt': 'world',
 125       'zipfile/readme': 'test zip',
 126       'zipfile/dir/file1': 'contents',
 127       'zipfile/dir/file2': 'more contents'
 128     }
 129
 130     string = _ZipFromFiles(files)
 131
 132     test_files = {
 133       'test_owner': {
 134         'changing-repo': {
 135           'commits': {
 136             'HEAD': make_sha_json(FAKE_HASH)
 137           },
 138           'zipball': string
 139         }
 140       }
 141     }
 142
 143     test_file_system = TestFileSystem(test_files)
 144
 145     gfs = GithubFileSystem.ForTest(
 146         'changing-repo', FakeURLFSFetcher.Create(test_file_system), path='')
 147
 148     gfs.Refresh().Get()
 149     initial_dir_read = sorted(gfs.ReadSingle('').Get())
 150     initial_file_read = gfs.ReadSingle('dir/file1').Get()
 151
 152     # Change the underlying data.
 153     files['zipfile/hello.txt'] = 'hello world again'
 154     files['zipfile/new-file'] = 'a new file'
 155     files['zipfile/dir/file1'] = 'changed file'
 156     test_files['test_owner']['changing-repo']['zipball'] = _ZipFromFiles(files)
 157     test_files['test_owner']['changing-repo']['commits']['HEAD'] = (
 158         make_sha_json(FAKE_HASH + 'hash'))
 159
 160     # Check that changes have not effected the file system yet.
 161     self.assertEqual(initial_dir_read, sorted(gfs.ReadSingle('').Get()))
 162     self.assertEqual(initial_file_read, gfs.ReadSingle('dir/file1').Get())
 163
 164     gfs.Refresh().Get()
 165
 166     # Check that the changes have effected the file system.
 167     self.assertTrue('new-file' in gfs.ReadSingle('').Get())
 168     self.assertEqual(files['zipfile/dir/file1'],
 169                      gfs.ReadSingle('dir/file1').Get())
 170
 171
 172 if __name__ == '__main__':
 173   unittest.main()