lib/test_vstorage.py

   1 #!/usr/bin/python
   2 # -*- coding: utf-8 -*-
   3 #
   4 # This file is part of FNP-Redakcja, licensed under GNU Affero GPLv3 or later.
   5 # Copyright © Fundacja Nowoczesna Polska. See NOTICE for more information.
   6 #
   7
   8 import os
   9 import tempfile
  10 from nose.tools import *
  11 from nose.core import runmodule
  12
  13 import vstorage
  14
  15 NULL_PARENT = -1
  16
  17
  18 def clear_directory(top):
  19     for root, dirs, files in os.walk(top, topdown=False):
  20         for name in files:
  21             os.remove(os.path.join(root, name))
  22         for name in dirs:
  23             os.rmdir(os.path.join(root, name))
  24     try:
  25         os.removedirs(top)
  26     except OSError:
  27         pass
  28
  29
  30 class TestVersionedStorage(object):
  31     def setUp(self):
  32         self.repo_path = tempfile.mkdtemp()
  33         self.repo = vstorage.VersionedStorage(self.repo_path)
  34
  35     def tearDown(self):
  36         clear_directory(self.repo_path)
  37
  38     def test_save_text(self):
  39         text = u"test text"
  40         title = u"test title"
  41         author = u"test author"
  42         comment = u"test comment"
  43
  44         self.repo.save_text(
  45             title=title,
  46             text=text,
  47             author=author,
  48             comment=comment,
  49             parent=NULL_PARENT,
  50         )
  51
  52         saved = self.repo.open_page(title).read()
  53         assert_equal(saved, text)
  54
  55     def test_save_text_noparent(self):
  56         text = u"test text"
  57         title = u"test title"
  58         author = u"test author"
  59         comment = u"test comment"
  60
  61         self.repo.save_text(title=title,
  62                     text=text, author=author,
  63                     comment=comment, parent=None)
  64
  65         saved = self.repo.open_page(title).read()
  66         assert_equal(saved, text)
  67
  68     def test_save_merge_no_conflict(self):
  69         text = u"test\ntext"
  70         title = u"test title"
  71         author = u"test author"
  72         comment = u"test comment"
  73         self.repo.save_text(title=title,
  74                     text=text, author=author,
  75                     comment=comment, parent=NULL_PARENT)
  76         self.repo.save_text(title=title,
  77                     text=text, author=author,
  78                     comment=comment, parent=NULL_PARENT)
  79         saved = self.repo.open_page(title).read()
  80         assert_equal(saved, text)
  81
  82     def test_save_merge_line_conflict(self):
  83         text = u"test\ntest\n"
  84         text1 = u"test\ntext\n"
  85         text2 = u"text\ntest\n"
  86         title = u"test title"
  87         author = u"test author"
  88         comment = u"test comment"
  89
  90         self.repo.save_text(title=title,
  91                     text=text, author=author,
  92                     comment=comment, parent=NULL_PARENT)
  93
  94         self.repo.save_text(title=title,
  95                     text=text1, author=author,
  96                     comment=comment, parent=0)
  97
  98         self.repo.save_text(title=title,
  99                     text=text2, author=author,
 100                     comment=comment, parent=0)
 101
 102         saved = self.repo.open_page(title).read()
 103
 104         # Other conflict markers placement can also be correct
 105         assert_equal(saved, u'''\
 106 text
 107 test
 108 <<<<<<< local
 109 =======
 110 text
 111 >>>>>>> other
 112 ''')
 113
 114     def test_delete(self):
 115         text = u"text test"
 116         title = u"test title"
 117         author = u"test author"
 118         comment = u"test comment"
 119         self.repo.save_text(title=title,
 120                     text=text, author=author,
 121                     comment=comment, parent=NULL_PARENT)
 122
 123         assert title in self.repo
 124
 125         self.repo.delete_page(title, author, comment)
 126
 127         assert title not in self.repo
 128
 129     @raises(vstorage.DocumentNotFound)
 130     def test_document_not_found(self):
 131         self.repo.open_page(u'unknown entity')
 132
 133     def test_open_existing_repository(self):
 134         self.repo.save_text(title=u'Python!', text=u'ham and spam')
 135         current_repo_revision = self.repo.repo_revision()
 136         same_repo = vstorage.VersionedStorage(self.repo_path)
 137         assert_equal(same_repo.repo_revision(), current_repo_revision)
 138
 139     def test_history(self):
 140         COMMITS = [
 141             {"author": "bunny", "text":"1", "comment": "Oh yeah!"},
 142             {"author": "frank", "text":"2", "comment": "Second is the best!"},
 143             {"text":"3", "comment": "Third"},
 144             {"author": "welma", "text":"4", "comment": "Fourth"},
 145         ]
 146
 147         for commit in COMMITS:
 148             self.repo.save_text(title=u"Sample", **commit)
 149
 150         for n, entry in enumerate(reversed(list(self.repo.page_history(u"Sample")))):
 151             assert_equal(entry["version"], n)
 152             assert_equal(entry["author"], COMMITS[n].get("author", "anonymous"))
 153             assert_equal(entry["description"], COMMITS[n]["comment"])
 154             assert_equal(entry["tag"], [])
 155
 156
 157 class TestVSTags(object):
 158
 159     TITLE_1 = "Sample"
 160
 161     COMMITS = [
 162         {"author": "bunny", "text":"1", "comment": "Oh yeah!"},
 163         {"author": "frank", "text":"2", "comment": "Second is the best!"},
 164         {"text":"3", "comment": "Third"},
 165         {"author": "welma", "text":"4", "comment": "Fourth"},
 166     ]
 167
 168     def setUp(self):
 169         self.repo_path = tempfile.mkdtemp()
 170         self.repo = vstorage.VersionedStorage(self.repo_path)
 171
 172         # generate some history
 173         for commit in self.COMMITS:
 174             self.repo.save_text(title=u"Sample", **commit)
 175
 176         # verify
 177         for n, entry in enumerate(reversed(list(self.repo.page_history(self.TITLE_1)))):
 178             assert_equal(entry["tag"], [])
 179
 180     def tearDown(self):
 181         clear_directory(self.repo_path)
 182
 183     def test_add_tag(self):
 184         TAG_USER = "mike_the_tagger"
 185         TAG_NAME = "production"
 186         TAG_VERSION = 2
 187
 188         # Add tag
 189         self.repo.add_page_tag(self.TITLE_1, TAG_VERSION, TAG_NAME, TAG_USER)
 190
 191         # check history again
 192         history = list(self.repo.page_history(self.TITLE_1))
 193         for entry in reversed(history):
 194             if entry["version"] == TAG_VERSION:
 195                 assert_equal(entry["tag"], [TAG_NAME])
 196             else:
 197                 assert_equal(entry["tag"], [])
 198
 199     def test_add_many_tags(self):
 200         TAG_USER = "mike_the_tagger"
 201         tags = [
 202             (2, "production", "mike"),
 203             (2, "finished", "jeremy"),
 204             (0, "original", "jeremy"),
 205         ]
 206
 207         for rev, name, user in tags:
 208             self.repo.add_page_tag(self.TITLE_1, rev, name, user)
 209
 210         # check history again
 211         history = list(self.repo.page_history(self.TITLE_1))
 212         for entry in reversed(history):
 213             expected = [tag[1] for tag in tags if tag[0] == entry["version"]]
 214             assert_equal(set(entry["tag"]), set(expected))
 215
 216
 217 if __name__ == '__main__':
 218     runmodule()