fuzzyworld-tools / files / tests / test_wp_import.py
test_wp_import.py
0e603cf7 2026-10-31 Pavel Safronov | 1 | """Tests for lib/wp_import.py. Run: python3 -m unittest discover tests |
| 2 | ||
| 3 | Every post, file and comment here is made up, in a temporary folder. No network: the API data is written | |
| 4 | straight into the importer's cache folder, and rsync and HEAD requests are replaced by fakes. | |
| 5 | """ | |
| 6 | import hashlib | |
| 7 | import json | |
| 8 | import os | |
| 9 | import re | |
| 10 | import sys | |
| 11 | import tempfile | |
| 12 | import unittest | |
| 13 | import urllib.error | |
| 14 | ||
| 15 | HERE = os.path.dirname(os.path.abspath(__file__)) | |
| 16 | sys.path.insert(0, os.path.join(HERE, "..", "lib")) | |
| 17 | import wp_import as wp # noqa: E402 | |
| 18 | ||
| 19 | blog = wp.blog | |
| 20 | needs_markdown = unittest.skipUnless(blog.markdown, "Python-Markdown is not installed (see FW_PYLIB)") | |
| 21 | ||
| 22 | WP_URL = "https://old.example.org" | |
| 23 | VOCAB = ("river stone lantern copper window harbor violet engine paper meadow signal orchard thunder velvet " | |
| 24 | "compass ladder whistle marble garden anchor falcon mirror candle bridge pebble saddle winter ribbon " | |
| 25 | "kettle summit shadow timber walnut yonder zephyr button carpet dragon ember feather glacier").split() | |
| 26 | ||
| 27 | ||
| 28 | def words(n, seed=0): | |
| 29 | """n made-up words, different for each seed.""" | |
| 30 | return " ".join(VOCAB[(i * (seed + 3) + seed * 7) % len(VOCAB)] + ("" if i % 5 else str(seed)) for i in range(n)) | |
| 31 | ||
| 32 | ||
| 33 | def md(html_text): | |
| 34 | return wp.html_to_markdown(html_text) | |
| 35 | ||
| 36 | ||
| 37 | class Converter(unittest.TestCase): | |
| 38 | def test_wordpress_additions_and_private_comments_removed(self): | |
| 39 | text = md('<div id="bsf_rt_marker"></div><span class="span-reading-time rt-reading-time"><span class="rt-label">' | |
| 40 | '[Reading time: 3 minutes]</span></span><p><!-- SECRET-CUT [x](y) --></p><p>Visible<span id="more-5"></span> text.</p>' | |
| 41 | '<p><!--more--></p><p>After</p><script async src="https://x.example/p.js"></script>') | |
| 42 | self.assertNotIn("SECRET-CUT", text) | |
| 43 | self.assertNotIn("Reading time", text) | |
| 44 | self.assertNotIn("script", text) | |
| 45 | self.assertEqual(text, "Visible text.\n\n<!--more-->\n\nAfter") | |
| 46 | ||
| 47 | def test_line_breaks_and_paragraphs(self): | |
| 48 | self.assertEqual(md("<p>one<br />\ntwo<br />\nthree</p>\n<p>next</p>"), "one\ntwo\nthree\n\nnext") | |
| 49 | # two <br> in a row: the second stays a literal <br> so the count of <br> is kept | |
| 50 | out = md("<p>one<br /><br />two</p>") | |
| 51 | self.assertEqual(out.count("\n") + out.count("<br>"), 2) | |
| 52 | self.assertEqual(md("<p>x<br /></p>"), "x") | |
| 53 | ||
| 54 | def test_inline_markup(self): | |
| 55 | self.assertEqual(md('<p><em>a</em> <strong>b</strong> <i>c</i> <code>d</code> <del>e</del> ' | |
| 56 | '<a href="https://old.example.org/x/y/">link</a></p>'), | |
| 57 | "*a* **b** *c* `d` <del>e</del> [link](https://old.example.org/x/y/)") | |
| 58 | self.assertEqual(md("<p><em>line one<br />line two</em></p>"), "*line one*\n*line two*") | |
| 59 | self.assertEqual(md('<p><a href="https://x.example/a b(1)">t</a> <img src="/u/a.png" alt="An [alt]"></p>'), | |
| 60 | "[t](https://x.example/a%20b%281%29) ") | |
| 61 | ||
| 62 | def test_special_characters_are_escaped(self): | |
| 63 | self.assertEqual(md("<p>2 * 3 < 5 & _x_ snake_case</p>"), "2 \\* 3 < 5 & \\_x\\_ snake_case") | |
| 64 | self.assertEqual(md("<p># not a heading<br />- not a list<br />1. not numbered<br />> not a quote</p>"), | |
| 65 | "\\# not a heading\n\\- not a list\n1\\. not numbered\n\\> not a quote") | |
| 66 | self.assertEqual(md("<p>&amp; stays</p>"), "&amp; stays") | |
| 67 | ||
| 68 | def test_blocks(self): | |
| 69 | out = md("<h2>Top</h2><h3>Mid</h3><h4>Low</h4><hr class='x'><blockquote><p>q1</p><p>q2</p></blockquote>" | |
| 70 | "<ul><li>a<ul><li>nested</li></ul></li><li>b</li></ul><ol><li>x</li><li>y</li></ol>" | |
| 71 | "<pre><code class=\"language-python\">print('hi')\n```\n</code></pre>") | |
| 72 | self.assertIn("### Top\n\n### Mid\n\n#### Low\n\n---\n\n> q1\n>\n> q2", out) # no ## heading: it would end the body | |
| 73 | self.assertIn("- a\n - nested\n- b", out) | |
| 74 | self.assertIn("1. x\n2. y", out) | |
| 75 | self.assertIn("````python\nprint('hi')\n```\n````", out) | |
| 76 | ||
| 77 | def test_embeds_become_links(self): | |
| 78 | out = md('<div class="jetpack-video-wrapper"><span class="embed-youtube"><iframe class="youtube-player" ' | |
| 79 | 'src="https://www.youtube.com/embed/abc_123?version=3&rel=1"></iframe></span></div>' | |
| 80 | '<p><a data-flickr-embed="true" href="https://flickr.example/p/1" title="A photo">' | |
| 81 | '<img src="https://flickr.example/i.jpg" alt="A photo"></a><script async src="x.js"></script></p>' | |
| 82 | '<div class="tumblr-post" data-href="h"><a href="https://tumblr.example/post/1">https://tumblr.example/post/1</a></div>') | |
| 83 | self.assertIn("[Video: https://www.youtube.com/watch?v=abc_123](https://www.youtube.com/watch?v=abc_123)", out) | |
| 84 | self.assertIn("[A photo](https://flickr.example/p/1)", out) | |
| 85 | self.assertNotIn("flickr.example/i.jpg", out) | |
| 86 | self.assertIn("[https://tumblr.example/post/1](https://tumblr.example/post/1)", out) | |
| 87 | notes = [] | |
| 88 | wp.html_to_markdown('<iframe src="https://other.example/frame"></iframe>', notes) | |
| 89 | self.assertEqual(notes, ["iframe dropped: https://other.example/frame"]) | |
| 90 | ||
| 91 | def test_text_and_br_count_ignore_what_is_removed(self): | |
| 92 | h = '<span class="span-reading-time">Reading time 1</span><p>a<br>b <!-- cut --> c</p><p>d</p>' | |
| 93 | self.assertEqual(wp.norm_words(wp.html_text(h)), ["a", "b", "c", "d"]) | |
| 94 | self.assertEqual(wp.html_br_count(h), 1) | |
| 95 | self.assertEqual(wp.html_plain("<p>one<br />two</p><p>three & four</p>"), "one\ntwo\n\nthree & four") | |
| 96 | ||
| 97 | def test_normalized_words_ignore_typography(self): | |
| 98 | self.assertEqual(wp.norm_words("It’s “quoted” – and… done--ok"), wp.norm_words("It's \"quoted\" - and... done - ok")) | |
| 99 | self.assertEqual(wp.md_words("Some *text* with [a link](https://x.example/q_r) and  <!--more--> end"), | |
| 100 | ["some", "text", "with", "a", "link", "and", "end"]) | |
| 101 | ||
| 102 | @needs_markdown | |
| 103 | def test_round_trip_through_the_generator(self): | |
| 104 | h = ("<p>First line<br />\nsecond *line* with 2 * 3 and _under_.</p><h3>Head</h3><p><em>Emphasis</em> and " | |
| 105 | "<strong>strong</strong>, <a href=\"https://old.example.org/a/\">a link</a>.</p><ul><li>one</li><li>two</li></ul>" | |
| 106 | "<blockquote><p>quoted words</p></blockquote><p>1. not a list</p>") | |
| 107 | text = "# T\n\n## Writing\n\n" + md(h) | |
| 108 | out = blog.render_markdown(blog.extract_body(text), {"line_breaks": "br", "rewrite_hosts": [], "redirects": {}, | |
| 109 | "wp_ids": {}}) | |
| 110 | self.assertGreaterEqual(wp.similarity(wp.norm_words(wp.html_text(h)), wp.norm_words(blog.plain_text(out))), 0.99) | |
| 111 | self.assertEqual(len(re.findall(r"<br", out)), wp.html_br_count(h)) | |
| 112 | self.assertNotIn("<ol", out) | |
| 113 | ||
| 114 | ||
| 115 | def post_json(i, title, slug, date, html_text, tags=(), protected=False, status="publish"): | |
| 116 | y, m, d = date[:10].split("-") | |
| 117 | return {"id": i, "date": date, "date_gmt": date.replace("T09", "T16"), "slug": slug, "status": status, "type": "post", | |
| 118 | "link": f"{WP_URL}/{y}/{m}/{d}/{slug}/", "title": {"rendered": title}, | |
| 119 | "content": {"rendered": "" if protected else html_text, "protected": protected}, | |
| 120 | "tags": list(tags), "class_list": ["post-password-required"] if protected else []} | |
| 121 | ||
| 122 | ||
| 123 | def comment_json(i, post, author, text, parent=0, status="approved"): | |
| 124 | return {"id": i, "post": post, "parent": parent, "author_name": author, "author_url": "https://spam.example/u", | |
| 125 | "author_email": "secret@example.org", "author_avatar_urls": {"24": "https://gravatar.example/a"}, | |
| 126 | "date": "2018-04-01T10:00:00", "status": status, "type": "comment", "content": {"rendered": f"<p>{text}</p>"}} | |
| 127 | ||
| 128 | ||
| 129 | def _sha(path): | |
| 130 | with open(path, "rb") as f: | |
| 131 | return hashlib.sha1(f.read()).hexdigest() | |
| 132 | ||
| 133 | ||
| 134 | class Site: | |
| 135 | """A made-up blog folder with a made-up WordPress API cache.""" | |
| 136 | ||
| 137 | def __init__(self, test, config=None): | |
| 138 | self.test = test | |
| 139 | self.dir = tempfile.mkdtemp() | |
| 140 | test.addCleanup(lambda: __import__("shutil").rmtree(self.dir, ignore_errors=True)) | |
| 141 | self.src = os.path.join(self.dir, "src") | |
| 142 | os.makedirs(self.src) | |
| 143 | self.config_path = os.path.join(self.dir, "fw-blog.json") | |
| 144 | self.config = {"title": "Test", "base_url": "https://new.example.org", "source": "src", "line_breaks": "br", | |
| 145 | "author": "A. Writer"} | |
| 146 | self.config.update(config or {}) | |
| 147 | self.save_config() | |
| 148 | self.api = os.path.join(self.src, "wordpress", "_api") | |
| 149 | self.fetch_calls = [] | |
| 150 | ||
| 151 | def save_config(self): | |
| 152 | with open(self.config_path, "w") as f: | |
| 153 | json.dump(self.config, f) | |
| 154 | ||
| 155 | def file(self, rel, text): | |
| 156 | path = os.path.join(self.src, rel) | |
| 157 | os.makedirs(os.path.dirname(path), exist_ok=True) | |
| 158 | with open(path, "w", encoding="utf-8") as f: | |
| 159 | f.write(text) | |
| 160 | ||
| 161 | def read(self, rel): | |
| 162 | with open(os.path.join(self.src, rel), encoding="utf-8") as f: | |
| 163 | return f.read() | |
| 164 | ||
| 165 | def raw_config(self): | |
| 166 | with open(self.config_path) as f: | |
| 167 | return json.load(f) | |
| 168 | ||
| 169 | def api_data(self, posts=(), pages=(), comments=(), tags=(), media=()): | |
| 170 | os.makedirs(self.api, exist_ok=True) | |
| 171 | for name, data in (("posts", posts), ("pages", pages), ("comments", comments), ("tags", tags), ("media", media)): | |
| 172 | with open(os.path.join(self.api, name + ".json"), "w") as f: | |
| 173 | json.dump(list(data), f) | |
| 174 | ||
| 175 | def snapshot(self): | |
| 176 | """{relative path: hash} of every file under the folder (config included).""" | |
| 177 | out = {} | |
| 178 | for root, _, names in os.walk(self.dir): | |
| 179 | for n in names: | |
| 180 | p = os.path.join(root, n) | |
| 181 | out[os.path.relpath(p, self.dir)] = _sha(p) | |
| 182 | return out | |
| 183 | ||
| 184 | def run(self, **kw): | |
| 185 | kw.setdefault("no_media", True) | |
| 186 | kw.setdefault("say", lambda s: None) | |
| 187 | return wp.run_import(self.config_path, WP_URL, **kw) | |
| 188 | ||
| 189 | ||
| 190 | def standard_site(test, **config): | |
| 191 | """Two matched-or-not public posts, one protected post, one page, one tag, comments.""" | |
| 192 | t = Site(test, config) | |
| 193 | first_body = words(60, 1) | |
| 194 | second_body = words(60, 2) | |
| 195 | t.file("2018-03-02-first-post.md", f"# First Post\n\n## Writing\n\n{first_body}\n\n## Spelling\n\nSECRET-SPELLING\n") | |
| 196 | t.file("2018-03-02-first-post (conflicted copy).md", f"# First Post\n\n## Writing\n\n{first_body}\n") | |
| 197 | t.file("2017-01-01-unrelated.md", f"# Unrelated\n\n## Writing\n\n{words(60, 9)}\n") | |
| 198 | t.api_data( | |
| 199 | posts=[post_json(11, "First Post", "first-post", "2018-03-02T09:15:00", f"<p>{first_body}</p>", tags=[5]), | |
| 200 | post_json(12, "Second – Post", "second-post", "2018-04-01T09:30:00", | |
| 201 | f"<div id=\"bsf_rt_marker\"></div><p>{second_body}<br />\nsecond line</p><p><!--more--></p><p>tail text</p>"), | |
| 202 | post_json(13, "Locked", "locked", "2018-05-01T09:00:00", "", protected=True)], | |
| 203 | pages=[{"id": 21, "slug": "highlights", "status": "publish", "link": WP_URL + "/", "title": {"rendered": "Highlights"}, | |
| 204 | "content": {"rendered": '<p>Front <a href="https://old.example.org/2018/04/01/second-post/">page</a>.</p>', "protected": False}}, | |
| 205 | {"id": 22, "slug": "blog", "status": "publish", "link": WP_URL + "/blog/", "title": {"rendered": "Blog"}, | |
| 206 | "content": {"rendered": "<p>generated</p>", "protected": False}}], | |
| 207 | comments=[comment_json(1, 12, "Pat", "Nice & good<br />really"), comment_json(2, 12, "Sam", "Reply", parent=1), | |
| 208 | comment_json(3, 13, "Eve", "On the locked post"), comment_json(4, 12, "Bot", "pending", status="hold")], | |
| 209 | tags=[{"id": 5, "slug": "fiction", "name": "Fiction", "link": WP_URL + "/tag/fiction/"}]) | |
| 210 | return t | |
| 211 | ||
| 212 | ||
| 213 | class Matching(unittest.TestCase): | |
| 214 | def files(self, *specs): | |
| 215 | import datetime | |
| 216 | return [{"rel": rel, "words": wp.norm_words(text), "title_words": wp.norm_words(title), | |
| 217 | "date": datetime.datetime.fromisoformat(date), "post": None} for rel, title, date, text in specs] | |
| 218 | ||
| 219 | def post(self, title, date, text): | |
| 220 | import datetime | |
| 221 | return {"title": title, "date": datetime.datetime.fromisoformat(date), "words": wp.norm_words(text), "path": "/" + title} | |
| 222 | ||
| 223 | def test_threshold_each_file_once_and_double_claims(self): | |
| 224 | body = words(80, 1) | |
| 225 | files = self.files(("a.md", "Same", "2018-01-01", body), ("far.md", "Other", "2018-01-01", words(80, 4))) | |
| 226 | near = body.replace("river", "rivers", 1) # one word differs: well above 0.95 | |
| 227 | posts = [self.post("Same", "2018-02-01", body), self.post("Copy", "2018-02-02", near), | |
| 228 | self.post("Far", "2018-02-03", words(80, 6))] | |
| 229 | assigned, best, conflicts = wp.match_posts(posts, files) | |
| 230 | self.assertEqual(sorted(assigned), [0]) # exact match wins, equal title; the copy gets nothing | |
| 231 | self.assertEqual(assigned[0][1], 0) | |
| 232 | self.assertTrue(any("claimed by" in c and "a.md" in c for c in conflicts)) | |
| 233 | self.assertNotIn(2, assigned) # no similar file | |
| 234 | files = self.files(("a.md", "Same", "2018-01-01", body)) | |
| 235 | half = " ".join(body.split()[:60]) + " " + words(20, 7) # about 75 percent similar | |
| 236 | assigned, best, _ = wp.match_posts([self.post("Same", "2018-02-01", half)], files) | |
| 237 | self.assertEqual(assigned, {}) | |
| 238 | self.assertTrue(0.5 < best[0][0][0] < wp.MATCH_THRESHOLD) # the score is still reported | |
| 239 | ||
| 240 | def test_ties_go_to_the_equal_title_then_the_earlier_date(self): | |
| 241 | body = words(50, 2) | |
| 242 | files = self.files(("x.md", "Different title", "2018-01-01", body), ("y.md", "The Title", "2018-01-05", body), | |
| 243 | ("z.md", "The Title", "2018-03-01", body)) | |
| 244 | assigned, _, conflicts = wp.match_posts([self.post("The Title", "2018-02-01", body)], files) | |
| 245 | self.assertEqual(files[assigned[0][1]]["rel"], "y.md") # equal title, dated before publication, nearest | |
| 246 | self.assertTrue(any("matches several files" in c for c in conflicts)) | |
| 247 | ||
| 248 | ||
| 249 | class Fetching(unittest.TestCase): | |
| 250 | def test_pagination_cache_and_refresh(self): | |
| 251 | tmp = tempfile.mkdtemp() | |
| 252 | self.addCleanup(lambda: __import__("shutil").rmtree(tmp, ignore_errors=True)) | |
| 253 | calls = [] | |
| 254 | ||
| 255 | def get_json(url): | |
| 256 | calls.append(url) | |
| 257 | page = int(re.search(r"[?&]page=(\d+)", url).group(1)) | |
| 258 | if page > 2: | |
| 259 | raise urllib.error.HTTPError(url, 400, "bad page", {}, None) | |
| 260 | return [{"id": page * 1000 + i} for i in range(100 if page == 1 else 3)], 2 | |
| 261 | ||
| 262 | f = wp.Fetcher(WP_URL, os.path.join(tmp, "_api"), get_json=get_json) | |
| 263 | self.assertEqual(len(f.collection("posts")), 103) | |
| 264 | self.assertEqual(len(calls), 2) | |
| 265 | self.assertIn("per_page=100", calls[0]) | |
| 266 | wp.Fetcher(WP_URL, os.path.join(tmp, "_api"), get_json=get_json).collection("posts") | |
| 267 | self.assertEqual(len(calls), 2) # cache used | |
| 268 | wp.Fetcher(WP_URL, os.path.join(tmp, "_api"), refresh=True, get_json=get_json).collection("posts") | |
| 269 | self.assertEqual(len(calls), 4) | |
| 270 | legacy = os.path.join(tmp, "legacy") | |
| 271 | os.makedirs(legacy) | |
| 272 | for name in ("p1.json", "p2.json"): # the development cache's names | |
| 273 | with open(os.path.join(legacy, name), "w") as fh: | |
| 274 | json.dump([{"id": name}], fh) | |
| 275 | self.assertEqual(wp.Fetcher(WP_URL, legacy).collection("posts"), [{"id": "p1.json"}, {"id": "p2.json"}]) | |
| 276 | ||
| 277 | ||
| 278 | class Import(unittest.TestCase): | |
| 279 | def test_dry_run_writes_only_api_and_report(self): | |
| 280 | t = standard_site(self) | |
| 281 | before = t.snapshot() | |
| 282 | report = t.run(dry_run=True) | |
| 283 | after = t.snapshot() | |
| 284 | new = sorted(set(after) - set(before)) | |
| 285 | self.assertEqual(new, ["src/wordpress/_report.txt"]) | |
| 286 | self.assertEqual({k: v for k, v in before.items() if k in after and after[k] != v}, {}) | |
| 287 | self.assertIn('"Locked"', report) # skipped posts are listed | |
| 288 | self.assertIn("password-protected", report) | |
| 289 | self.assertIn("DRY RUN", report) | |
| 290 | self.assertRegex(report, r"1\.0000 2018-03-02-first-post\.md <- \"First Post\"") | |
| 291 | self.assertNotIn("conflicted copy", report) | |
| 292 | self.assertIn("wordpress/2018-04-01-second-post.md", report) | |
| 293 | ||
| 294 | def test_import_writes_files_comments_and_config(self): | |
| 295 | t = standard_site(self) | |
| 296 | t.run() | |
| 297 | second = t.read("wordpress/2018-04-01-second-post.md") | |
| 298 | self.assertEqual(second.split("\n")[0], "# Second – Post") | |
| 299 | self.assertIn("## Writing\n\n", second) | |
| 300 | self.assertIn("second line", second) | |
| 301 | self.assertNotIn("bsf_rt", second) | |
| 302 | self.assertIn("<!--more-->", second) | |
| 303 | self.assertTrue(t.read("pages/highlights.md").startswith( | |
| 304 | "---\npublished: true\n---\n# Highlights\n\nFront [page](https://old.example.org/")) # pages are opt-in | |
| 305 | self.assertFalse(os.path.exists(os.path.join(t.src, "pages", "blog.md"))) | |
| 306 | self.assertFalse(os.path.exists(os.path.join(t.src, "wordpress", "2018-03-02-first-post.md"))) # matched: the source file is used | |
| 307 | self.assertFalse(os.path.exists(os.path.join(t.src, "comments.json"))) # relative to the config, not to `source` | |
| 308 | with open(os.path.join(t.dir, "comments.json"), encoding="utf-8") as f: | |
| 309 | comments = json.load(f) | |
| 310 | self.assertEqual(list(comments), ["/2018/04/01/second-post/"]) | |
| 311 | self.assertEqual(comments["/2018/04/01/second-post/"][0], | |
| 312 | {"id": 1, "parent": 0, "author": "Pat", "date": "2018-04-01T10:00:00", "text": "Nice & good\nreally"}) | |
| 313 | self.assertEqual(sorted(comments["/2018/04/01/second-post/"][1]), ["author", "date", "id", "parent", "text"]) | |
| 314 | raw = t.raw_config() | |
| 315 | self.assertNotIn("secret", json.dumps(comments)) | |
| 316 | self.assertNotIn("spam.example", json.dumps(comments)) | |
| 317 | self.assertEqual(raw["posts"]["2018-03-02-first-post.md"], { | |
| 318 | "published": True, "date": "2018-03-02T09:15:00", "slug": "first-post", "wp_id": 11, "tags": ["fiction"]}) | |
| 319 | self.assertEqual(raw["posts"]["wordpress/2018-04-01-second-post.md"]["wp_id"], 12) | |
| 320 | self.assertEqual(raw["tags"], {"fiction": "Fiction"}) | |
| 321 | self.assertEqual(raw["wp_ids"], {"11": "/2018/03/02/first-post/", "12": "/2018/04/01/second-post/", "21": "/"}) | |
| 322 | self.assertEqual((raw["pages_dir"], raw["home"], raw["comments"], raw["timezone"]), | |
| 323 | ("pages", "highlights", "comments.json", "America/Los_Angeles")) | |
| 324 | self.assertNotIn("media", raw) # run with no_media | |
| 325 | self.assertIn("wordpress/*.md", raw["include"]) | |
| 326 | self.assertIn(WP_URL, raw["rewrite_hosts"]) | |
| 327 | self.assertNotIn("locked", json.dumps(raw).lower()) # protected: no entries anywhere | |
| 328 | self.assertNotIn("On the locked post", json.dumps(comments)) | |
| 329 | blog.load_config(t.config_path) # still a valid config | |
| 330 | ||
| 331 | def test_idempotent_and_never_overwrites_without_force(self): | |
| 332 | t = standard_site(self) | |
| 333 | t.run() | |
| 334 | snap = t.snapshot() | |
| 335 | t.run() | |
| 336 | self.assertEqual(t.snapshot(), snap) | |
| 337 | edited = t.read("wordpress/2018-04-01-second-post.md") + "\nMy own edit.\n" | |
| 338 | t.file("wordpress/2018-04-01-second-post.md", edited) | |
| 339 | raw = t.raw_config() | |
| 340 | raw["posts"]["wordpress/2018-04-01-second-post.md"]["slug"] = "mine" | |
| 341 | t.config = raw | |
| 342 | t.save_config() | |
| 343 | snap = t.snapshot() | |
| 344 | report = t.run() | |
| 345 | changed = {k for k, v in t.snapshot().items() if snap.get(k) != v} | |
| 346 | self.assertEqual(changed, {"src/wordpress/_report.txt"}) # only the report (it now lists the kept file) | |
| 347 | self.assertIn("Files kept because they differ", report) | |
| 348 | self.assertIn(" wordpress/2018-04-01-second-post.md", report) | |
| 349 | self.assertIn("kept existing slug", report) | |
| 350 | t.run(force=True) | |
| 351 | self.assertNotIn("My own edit.", t.read("wordpress/2018-04-01-second-post.md")) | |
| 352 | self.assertEqual(t.raw_config()["posts"]["wordpress/2018-04-01-second-post.md"]["slug"], "second-post") | |
| 353 | ||
| 354 | def test_existing_config_entries_and_body_until_are_kept(self): | |
| 355 | t = standard_site(self, posts={"2018-03-02-first-post.md": {"published": False, "body_until": "Info", "title": "Mine"}}, | |
| 356 | tags={"fiction": "My fiction"}, home="elsewhere") | |
| 357 | t.run() | |
| 358 | raw = t.raw_config() | |
| 359 | entry = raw["posts"]["2018-03-02-first-post.md"] | |
| 360 | self.assertEqual((entry["published"], entry["body_until"], entry["title"]), (False, "Info", "Mine")) | |
| 361 | self.assertEqual(entry["wp_id"], 11) # missing keys are added | |
| 362 | self.assertEqual((raw["tags"], raw["home"]), ({"fiction": "My fiction"}, "elsewhere")) | |
| 363 | ||
| 364 | def test_second_source_file_with_same_text_is_flagged_not_used_twice(self): | |
| 365 | t = standard_site(self) | |
| 366 | os.remove(os.path.join(t.src, "2018-03-02-first-post (conflicted copy).md")) | |
| 367 | t.file("2018-03-03-first-post-again.md", t.read("2018-03-02-first-post.md")) | |
| 368 | report = t.run() | |
| 369 | self.assertIn("matches several files", report) | |
| 370 | self.assertEqual(sum(1 for line in report.split("\n") if re.match(r"\s+1\.0000", line)), 1) | |
| 371 | ||
| 372 | def test_title_difference_is_recorded(self): | |
| 373 | t = standard_site(self) | |
| 374 | t.file("2018-03-02-first-post.md", t.read("2018-03-02-first-post.md").replace("# First Post", "# Working Title")) | |
| 375 | t.run() | |
| 376 | self.assertEqual(t.raw_config()["posts"]["2018-03-02-first-post.md"]["title"], "First Post") | |
| 377 | ||
| 378 | def test_media_sync_command_is_restricted_and_inside_the_source(self): | |
| 379 | t = standard_site(self) | |
| 380 | calls = [] | |
| 381 | ||
| 382 | class Result: | |
| 383 | returncode, stderr = 0, "" | |
| 384 | ||
| 385 | def fake(cmd, **kw): | |
| 386 | calls.append(cmd) | |
| 387 | return Result() | |
| 388 | t.run(no_media=False, rsync=fake, media_from="dh:old.example.org/wp-content/uploads/") | |
| 389 | self.assertEqual(len(calls), 1) | |
| 390 | cmd = calls[0] | |
| 391 | self.assertEqual(cmd[0], "rsync") | |
| 392 | self.assertNotIn("--delete", cmd) | |
| 393 | self.assertEqual(cmd[-2], "dh:old.example.org/wp-content/uploads/") | |
| 394 | self.assertTrue(cmd[-1].endswith("/")) | |
| 395 | self.assertEqual(os.path.realpath(cmd[-1]), os.path.realpath(os.path.join(t.src, "media", "wp-content", "uploads"))) | |
| 396 | self.assertEqual(t.raw_config()["media"], "media") | |
| 397 | includes = [c for c in cmd if c.startswith("--include")] | |
| 398 | self.assertIn("--include=*.[jJ][pP][gG]", includes) | |
| 399 | self.assertIn("--exclude=*", cmd) | |
| 400 | self.assertEqual(cmd[cmd.index("--exclude=*") + 1:cmd.index("--exclude=*") + 2], [cmd[-2]]) | |
| 401 | for c in includes: | |
| 402 | self.assertNotIn("md", c.lower().replace("*", "")) # no Markdown, JSON or script patterns | |
| 403 | t.run(no_media=False, rsync=fake) | |
| 404 | self.assertEqual(calls[1][-2], "dh:old.example.org/wp-content/uploads/") # default derived from --wp | |
| 405 | n = len(calls) | |
| 406 | t.run(dry_run=True, no_media=False, rsync=fake) | |
| 407 | self.assertEqual(len(calls), n) # a dry run never copies | |
| 408 | ||
| 409 | def test_probe_links_only_with_the_flag_and_records_301_targets(self): | |
| 410 | t = standard_site(self) | |
| 411 | t.api_data(posts=json.load(open(os.path.join(t.api, "posts.json"))) + [ | |
| 412 | post_json(14, "Linky", "linky", "2018-06-01T09:00:00", | |
| 413 | '<p><a href="https://old.example.org/2017/old-name/">old</a> <a href="/gone/">gone</a> ' | |
| 414 | '<a href="https://old.example.org/2018/04/01/second-post/">fine</a> <a href="https://elsewhere.example/x/">out</a> ' | |
| 415 | '<img src="https://old.example.org/wp-content/uploads/2018/06/missing.png"></p>')], | |
| 416 | pages=json.load(open(os.path.join(t.api, "pages.json"))), | |
| 417 | comments=json.load(open(os.path.join(t.api, "comments.json"))), | |
| 418 | tags=json.load(open(os.path.join(t.api, "tags.json")))) | |
| 419 | heads = [] | |
| 420 | ||
| 421 | def head(url): | |
| 422 | heads.append(url) | |
| 423 | return {WP_URL + "/2017/old-name/": (301, WP_URL + "/2018/03/02/first-post/")}.get(url, (404, None)) | |
| 424 | ||
| 425 | fetcher = wp.Fetcher(WP_URL, t.api, head=head) | |
| 426 | report = t.run(dry_run=True, fetcher=fetcher) | |
| 427 | self.assertEqual(heads, []) | |
| 428 | self.assertIn("unchecked (use --probe-links): /2017/old-name/", report) | |
| 429 | self.assertIn("missing media: /wp-content/uploads/2018/06/missing.png", report) | |
| 430 | self.assertNotIn("elsewhere.example", report) | |
| 431 | report = t.run(dry_run=True, probe=True, fetcher=fetcher) | |
| 432 | self.assertEqual(sorted(heads), [WP_URL + "/2017/old-name/", WP_URL + "/gone/"]) | |
| 433 | self.assertIn("redirect /2017/old-name/ -> /2018/03/02/first-post/", report) | |
| 434 | self.assertIn("/gone/: HTTP 404", report) | |
| 435 | self.assertEqual(t.raw_config().get("redirects"), None) # a dry run does not touch the config | |
| 436 | t.run(probe=True, fetcher=fetcher) | |
| 437 | self.assertEqual(t.raw_config()["redirects"], {"/2017/old-name/": "/2018/03/02/first-post/"}) | |
| 438 | self.assertEqual(len(heads), 2) # probes are cached in _api/probes.json | |
| 439 | ||
| 440 | def test_paths_cannot_leave_the_source_folder(self): | |
| 441 | t = standard_site(self) | |
| 442 | with self.assertRaises(wp.WpImportError): | |
| 443 | wp.write_file(t.src, "../escape.md", "x", False, False) | |
| 444 | self.assertEqual(wp.safe_slug("../../etc/passwd", 7), "etcpasswd") | |
| 445 | self.assertEqual(wp.safe_slug("...", 7), "post-7") | |
| 446 | ||
| 447 | ||
| 448 | def test_probe_skips_links_that_cannot_go_into_htaccess(self): | |
| 449 | heads = [] | |
| 450 | ||
| 451 | def head(url): | |
| 452 | heads.append(url) | |
| 453 | return {WP_URL + "/moved/": (301, WP_URL + "/new%0Aline/"), WP_URL + "/fine-old/": (301, WP_URL + "/fine-new/")}.get(url, (404, None)) | |
| 454 | ||
| 455 | t = Site(self) | |
| 456 | fetcher = wp.Fetcher(WP_URL, t.api, head=head) | |
| 457 | redirects, lines = wp.probe_links(fetcher, WP_URL, ["/fine-old/", "/with%20space/", "/line%0Abreak/", "/moved/"], | |
| 458 | set(), os.path.join(t.api, "probes.json"), True) | |
| 459 | self.assertEqual(redirects, {"/fine-old/": "/fine-new/"}) | |
| 460 | self.assertEqual(sorted(heads), [WP_URL + "/fine-old/", WP_URL + "/moved/"]) # unsafe links are not even requested | |
| 461 | report = "\n".join(lines) | |
| 462 | self.assertIn("with%20space", report) | |
| 463 | self.assertIn("line%0Abreak", report) | |
| 464 | self.assertIn("skipped", report) | |
| 465 | self.assertIn("/moved/ -> '/new%0Aline/': skipped", report) | |
| 466 | ||
| 467 | def test_comments_file_is_relative_to_the_config_folder_and_the_generator_reads_it(self): | |
| 468 | t = standard_site(self, source="src") | |
| 469 | t.run() | |
| 470 | self.assertTrue(os.path.isfile(os.path.join(t.dir, "comments.json"))) | |
| 471 | self.assertFalse(os.path.exists(os.path.join(t.src, "comments.json"))) | |
| 472 | if blog.markdown: | |
| 473 | out = os.path.join(t.dir, "out") | |
| 474 | blog.build(blog.load_config(t.config_path), out) | |
| 475 | with open(os.path.join(out, "2018/04/01/second-post/index.html"), encoding="utf-8") as f: | |
| 476 | self.assertIn("Nice & good", f.read()) | |
| 477 | ||
| 478 | def test_percent_encoded_slugs_get_readable_file_names_and_keep_their_url(self): | |
| 479 | self.assertEqual(wp.safe_slug("caf%c3%a9-night", 5), "café-night") | |
| 480 | self.assertEqual(wp.safe_slug("%2e%2e%2fetc", 5), "etc") | |
| 481 | self.assertEqual(wp.safe_slug("%e6%97%a5%e6%9c%ac", 5), "日本") | |
| 482 | t = Site(self) | |
| 483 | slug = "caf%c3%a9-night" | |
| 484 | t.api_data( | |
| 485 | posts=[post_json(31, "Café Night", slug, "2018-05-01T09:00:00", f"<p>{words(30, 4)}</p>")], | |
| 486 | pages=[{"id": 41, "slug": "d%c3%a9j%c3%a0", "status": "publish", "link": WP_URL + "/d%c3%a9j%c3%a0/", | |
| 487 | "title": {"rendered": "Déjà"}, "content": {"rendered": "<p>Page text.</p>", "protected": False}}]) | |
| 488 | t.run() | |
| 489 | self.assertTrue(os.path.isfile(os.path.join(t.src, "wordpress", "2018-05-01-café-night.md"))) | |
| 490 | raw = t.raw_config() | |
| 491 | self.assertEqual(raw["posts"]["wordpress/2018-05-01-café-night.md"]["slug"], slug) # the URL is what WordPress has | |
| 492 | self.assertEqual(raw["wp_ids"]["31"], f"/2018/05/01/{slug}/") | |
| 493 | page = t.read("pages/déjà.md") | |
| 494 | self.assertTrue(page.startswith("---\npublished: true\nslug: d%c3%a9j%c3%a0\n---\n")) | |
| 495 | blog.load_config(t.config_path) | |
| 496 | if blog.markdown: | |
| 497 | out = os.path.join(t.dir, "out") | |
| 498 | blog.build(blog.load_config(t.config_path), out) | |
| 499 | self.assertTrue(wp.output_has(out, f"/2018/05/01/{slug}/")) | |
| 500 | self.assertTrue(wp.output_has(out, "/d%c3%a9j%c3%a0/")) | |
| 501 | self.assertTrue(os.path.isfile(os.path.join(out, "2018/05/01/café-night/index.html"))) | |
| 502 | ||
| 503 | ||
| 504 | @needs_markdown | |
| 505 | class Check(unittest.TestCase): | |
| 506 | def build_checked(self, t, **kw): | |
| 507 | t.run() | |
| 508 | report, problems = wp.run_check(t.config_path, **kw) | |
| 509 | return report, problems | |
| 510 | ||
| 511 | def test_clean_import_passes(self): | |
| 512 | t = standard_site(self) | |
| 513 | report, problems = self.build_checked(t) | |
| 514 | self.assertEqual(problems, 0, report) | |
| 515 | self.assertIn("OK", report) | |
| 516 | ||
| 517 | def test_differences_are_listed(self): | |
| 518 | t = standard_site(self) | |
| 519 | t.run() | |
| 520 | text = t.read("2018-03-02-first-post.md") | |
| 521 | t.file("2018-03-02-first-post.md", text.replace(words(60, 1), " ".join(words(60, 1).split()[:50]) + " added words in the middle of it")) | |
| 522 | raw = t.raw_config() | |
| 523 | raw["redirects"] = {"/old/": "/does/not/exist/"} | |
| 524 | t.config = raw | |
| 525 | t.save_config() | |
| 526 | report, problems = wp.run_check(t.config_path) | |
| 527 | self.assertGreaterEqual(problems, 2) | |
| 528 | self.assertRegex(report, r"Posts below 0\.98 similarity[^\n]*: 1 ==\n 0\.\d+ /2018/03/02/first-post/") | |
| 529 | self.assertIn("/old/ -> /does/not/exist/", report) | |
| 530 | ||
| 531 | def test_br_counts_and_missing_urls(self): | |
| 532 | t = standard_site(self) | |
| 533 | t.run() | |
| 534 | raw = t.raw_config() | |
| 535 | raw["line_breaks"] = "markdown" | |
| 536 | t.config = raw | |
| 537 | t.save_config() | |
| 538 | report, problems = wp.run_check(t.config_path) | |
| 539 | self.assertIn("<br> counts that differ: 1", report) | |
| 540 | self.assertIn("WordPress 1, built 0", report) | |
| 541 | raw["posts"]["wordpress/2018-04-01-second-post.md"]["published"] = False | |
| 542 | t.config = raw | |
| 543 | t.save_config() | |
| 544 | report, _ = wp.run_check(t.config_path) | |
| 545 | self.assertIn("post /2018/04/01/second-post/", report) | |
| 546 | self.assertIn("Old URLs missing from the output: 1", report) | |
| 547 | ||
| 548 | def test_check_needs_the_cache(self): | |
| 549 | t = Site(self) | |
| 550 | with self.assertRaises(wp.WpImportError): | |
| 551 | wp.run_check(t.config_path) | |
| 552 | ||
| 553 | ||
| 554 | if __name__ == "__main__": | |
| 555 | unittest.main() |