Spaces:
Paused
Paused
| """/shift: posts of one channel copied into another - range, buttons, replies, links between the posts, undo.""" | |
| import asyncio | |
| from types import SimpleNamespace | |
| import pytest | |
| from telethon import functions, types | |
| from app.channelref import classify_ref, resolve_channel | |
| from app.common import UserError | |
| from app.handlers.shift import parse_shift_args | |
| from app.repost_engine import RepostOptions, plan_repost | |
| from app.shift_engine import delete_shift_copies, run_shift | |
| from app.tgutil import button_url | |
| from .dbutil import db_url_for, fresh_db | |
| from .fakes import FakeChannelClient, cb_btn, make_msg, markup, photo_media, service_msg, url_btn | |
| from .harness import App, norm | |
| from .test_repost_flow import fill | |
| SRC = SimpleNamespace(id=1, access_hash=5, title="Anime Source", username="srcch") | |
| def source_posts(): | |
| return { | |
| 1: make_msg(1, "Welcome", [types.MessageEntityBold(0, 7)]), | |
| 2: make_msg( | |
| 2, "Click below", | |
| markup=markup([url_btn("Download", "https://t.me/goku?start=AAA")], [cb_btn("👍 3", b"r")]), | |
| ), | |
| 3: make_msg(3, "Caption with link", [types.MessageEntityTextUrl(13, 4, "https://t.me/srcch/5")], media=photo_media(31)), | |
| 4: service_msg(4), | |
| 5: make_msg(5, "Fifth", reply_to=2, pinned=True), | |
| 6: make_msg(6, "", media=photo_media(61), grouped_id=60), | |
| 7: make_msg(7, "album caption", media=photo_media(62), grouped_id=60), | |
| 8: make_msg(8, "See https://t.me/srcch/1 and https://t.me/c/1/2 and https://t.me/elsewhere/9"), | |
| } | |
| def old_posts(): | |
| """Posts the destination already has, so that the copies get ids that differ from the originals'.""" | |
| return {i: make_msg(i, f"existing {i}", out=True) for i in range(1, 11)} | |
| def two_channels(src_msgs=None, dst_msgs=None): | |
| src = FakeChannelClient(source_posts() if src_msgs is None else src_msgs, username="srcch", title="Anime Source") | |
| dst = src.add_channel(FakeChannelClient(old_posts() if dst_msgs is None else dst_msgs, username="dstch", cid=2, title="Anime Backup")) | |
| return src, dst | |
| async def engine_setup(tmp_path, src, first=1, last=8): | |
| db = await fresh_db(db_url_for(tmp_path)) | |
| await db.save_channel(2, 6, "Anime Backup", "dstch", 1) | |
| dst_row = await db.get_channel(2) | |
| await db.create_shift("s1", src=SRC, dst_channel_id=2, first_id=first, last_id=last, user_id=1) | |
| return db, dst_row, await db.get_shift("s1") | |
| def reply_of(m): | |
| return m.reply_to.reply_to_msg_id if m.reply_to else None | |
| def run(coro): | |
| return asyncio.run(coro) | |
| # ------------------------------------------------------------------------------------- the command line | |
| def test_arguments_are_read_as_source_destination_and_an_optional_range(): | |
| def read(raw): | |
| *four, flags = parse_shift_args(raw) | |
| assert (flags.links, flags.profile) == (False, False) # no extras unless they are asked for | |
| return tuple(four) | |
| assert read("@a @b") == ("@a", "@b", None, None) | |
| assert read("@a @b 15 20") == ("@a", "@b", 15, 20) | |
| assert read("@a @b 20 15") == ("@a", "@b", 15, 20) # either order | |
| assert read("@a @b 15") == ("@a", "@b", 15, 15) # one id = just that post | |
| assert read("https://t.me/+AbCdEfGhIj -1001234567890") == ("https://t.me/+AbCdEfGhIj", "-1001234567890", None, None) | |
| for bad in ("", "@a", "@a @b x", "@a @b 0", "@a @b 1 2 3", "@a @b -5", "-c", "-c -all", "@a -c", "@a @b -c x"): | |
| with pytest.raises(ValueError): | |
| parse_shift_args(bad) | |
| def test_the_two_extras_can_be_given_separately_together_in_any_order_or_not_at_all(): | |
| def read(raw): | |
| src, dst, first, last, flags = parse_shift_args(raw) | |
| assert (src, dst) == ("@old", "@new") | |
| return (first, last), (flags.links, flags.profile) | |
| assert read("@old @new") == ((None, None), (False, False)) | |
| assert read("@old @new -c") == ((None, None), (True, False)) | |
| assert read("@old @new -all") == ((None, None), (False, True)) | |
| assert read("@old @new -c -all") == ((None, None), (True, True)) | |
| assert read("@old @new -all -c") == ((None, None), (True, True)) | |
| assert read("@old @new - all") == ((None, None), (False, True)) # typed with a space | |
| assert read("@old @new - c - all") == ((None, None), (True, True)) | |
| assert read("@old @new -ALL -C") == ((None, None), (True, True)) # case does not matter | |
| assert read("@old @new --all --c") == ((None, None), (True, True)) # a double dash is understood too | |
| assert read("@old @new 15 20 -c") == ((15, 20), (True, False)) # with a range | |
| assert read("-c @old @new -all 15 20") == ((15, 20), (True, True)) # in front, or between the other words | |
| assert read("@old -c @new 15 -all 20") == ((15, 20), (True, True)) | |
| assert read("@old @new -c -c") == ((None, None), (True, False)) # twice is the same as once | |
| def test_a_channel_id_is_not_mistaken_for_an_option_and_unknown_options_are_refused(): | |
| src, dst, first, last, flags = parse_shift_args("-1001234567890 -1009876543210 -c") | |
| assert (src, dst, flags.links) == ("-1001234567890", "-1009876543210", True) | |
| for bad in ("@a @b -x", "@a @b --everything", "@a @b -call"): | |
| with pytest.raises(ValueError, match="Unknown option"): | |
| parse_shift_args(bad) | |
| def test_every_kind_of_channel_reference_is_told_apart(): | |
| assert classify_ref("@srcch") == ("username", "srcch") | |
| assert classify_ref("https://t.me/srcch") == ("username", "srcch") | |
| assert classify_ref("https://t.me/srcch/15") == ("username", "srcch") | |
| assert classify_ref("-1001234567890") == ("id", 1234567890) | |
| assert classify_ref("https://t.me/c/1234567890/5") == ("id", 1234567890) | |
| assert classify_ref("https://t.me/+AbCdEfGhIjKl") == ("invite", "AbCdEfGhIjKl") | |
| assert classify_ref("t.me/joinchat/AbCdEfGhIjKl") == ("invite", "AbCdEfGhIjKl") | |
| assert classify_ref("tg://join?invite=AbCdEfGhIjKl") == ("invite", "AbCdEfGhIjKl") | |
| assert classify_ref("???") == (None, None) | |
| def test_channels_are_found_by_name_id_or_invite_link(tmp_path): | |
| async def go(): | |
| app = App(tmp_path) | |
| await app.start() | |
| src, dst = two_channels() | |
| app.tg.registry = src.registry | |
| app.tg.registry[1] = app.tg | |
| app.tg.username, app.tg.title = "srcch", "Anime Source" | |
| app.tg.add_channel(dst) | |
| await app.db.save_channel(1, 5, "Anime Source", "srcch", 1) # registered | |
| # registered: found in the database, whatever the spelling | |
| for text in ("@srcch", "SRCCH", "https://t.me/srcch/12", "1"): | |
| ref = await resolve_channel(app.ctx, text) | |
| assert (ref.id, ref.title, ref.registered) == (1, "Anime Source", True), text | |
| # not registered: asked from Telegram by name, by id (a channel the bot has met) or by invite link | |
| ref = await resolve_channel(app.ctx, "@dstch") | |
| assert (ref.id, ref.title, ref.username, ref.registered) == (2, "Anime Backup", "dstch", False) | |
| assert (await resolve_channel(app.ctx, "2")).id == 2 | |
| dst.invite_hash = "AbCdEfGhIjKl" | |
| assert (await resolve_channel(app.ctx, "https://t.me/+AbCdEfGhIjKl")).id == 2 | |
| # things that cannot be used say why | |
| for text, part in ( | |
| ("???", "not a channel I understand"), | |
| ("@nosuchchannel", "couldn't find a channel"), | |
| ("99", "don't know a channel with the id 99"), | |
| ("https://t.me/+ZzZzZzZzZz", "invite link doesn't work"), | |
| ): | |
| with pytest.raises(UserError, match=part): | |
| await resolve_channel(app.ctx, text) | |
| await app.db.close() | |
| run(go()) | |
| def test_a_group_or_a_user_is_not_a_channel(tmp_path): | |
| async def go(): | |
| app = App(tmp_path) | |
| await app.start() | |
| group = FakeChannelClient({}, username="agroup", cid=3) | |
| group._channel_obj = lambda: types.Channel(id=3, title="G", photo=types.ChatPhotoEmpty(), date=None, access_hash=1, username="agroup", megagroup=True) | |
| app.tg.add_channel(group) | |
| with pytest.raises(UserError, match="group, not a channel"): | |
| await resolve_channel(app.ctx, "@agroup") | |
| await app.db.close() | |
| run(go()) | |
| # ------------------------------------------------------------------------------------------ the engine | |
| def test_posts_arrive_in_the_other_channel_with_everything_they_had(tmp_path): | |
| async def go(): | |
| src, dst = two_channels() | |
| db, dst_row, shift = await engine_setup(tmp_path, src) | |
| res = await run_shift(src, db, shift, dst_row, 99, delay=0, settle_pause=0) | |
| assert (res.copied_units, res.copied_messages, res.failed, res.aborted) == (6, 7, [], None) | |
| new = {i: m for i, m in dst.msgs.items() if i > 10} | |
| assert list(new) == [11, 12, 13, 14, 15, 16, 17] # same order as the originals | |
| assert new[11].message == "Welcome" and isinstance(new[11].entities[0], types.MessageEntityBold) | |
| # the button post: link button kept, the other bot's 👍 button is not copied (it would be dead) | |
| rows = new[12].reply_markup.rows | |
| assert len(rows) == 1 and button_url(rows[0].buttons[0]) == "https://t.me/goku?start=AAA" and rows[0].buttons[0].text == "Download" | |
| assert new[13].media.photo.id == 31 | |
| # a reply answers the COPY of the post it answered; a pin is not copied | |
| assert reply_of(new[14]) == 12 and not new[14].pinned | |
| assert new[15].grouped_id and new[15].grouped_id == new[16].grouped_id and new[16].message == "album caption" | |
| # links to other source posts follow the content: backward ones at once, forward ones afterwards | |
| assert new[17].message == "See https://t.me/dstch/11 and https://t.me/dstch/12 and https://t.me/elsewhere/9" | |
| link = [e for e in new[13].entities if isinstance(e, types.MessageEntityTextUrl)][0] | |
| assert link.url == "https://t.me/dstch/14" and (link.offset, link.length) == (13, 4) | |
| assert (res.final.relinked, res.final.links) == (1, 1) | |
| # the source was only read | |
| assert sorted(src.msgs) == [1, 2, 3, 4, 5, 6, 7, 8] and src.msgs[5].pinned | |
| assert src.delete_requests == [] and src.edits == [] | |
| assert not any(isinstance(r, (functions.messages.SendMessageRequest, functions.messages.SendMediaRequest)) for r in src.requests) | |
| # recorded for My posts: only the destination has posts, exactly as they look there | |
| assert await db.shift_pairs("s1") == [(1, 11), (2, 12), (3, 13), (5, 14), (6, 15), (7, 16), (8, 17)] | |
| assert await db.sent_posts(1) == [] | |
| mine = {p.message_id: p for p in await db.sent_posts(2)} | |
| assert sorted(mine) == [11, 12, 13, 14, 15, 16, 17] and mine[12].source == "bot" | |
| assert mine[12].buttons == [[{"t": "Download", "u": "https://t.me/goku?start=AAA"}]] | |
| assert mine[13].entities[0]["u"] == "https://t.me/dstch/14" | |
| assert mine[17].text == new[17].message and mine[13].media_kind == "photo" | |
| await db.close() | |
| run(go()) | |
| def test_only_the_posts_in_the_range_are_copied(tmp_path): | |
| async def go(): | |
| src, dst = two_channels() | |
| db, dst_row, shift = await engine_setup(tmp_path, src, first=2, last=3) | |
| plan = await plan_repost(src, SRC, RepostOptions(), first_id=2, last_id=3) | |
| assert (plan.first, plan.last, plan.units, plan.partial) == (2, 3, 2, True) | |
| res = await run_shift(src, db, shift, dst_row, 99, delay=0, settle_pause=0) | |
| assert (res.copied_units, res.failed) == (2, []) | |
| assert [m.message for i, m in sorted(dst.msgs.items()) if i > 10] == ["Click below", "Caption with link"] | |
| await db.close() | |
| run(go()) | |
| def test_a_reply_to_a_post_outside_the_range_is_copied_without_the_reply(tmp_path): | |
| async def go(): | |
| src, dst = two_channels() | |
| db, dst_row, shift = await engine_setup(tmp_path, src, first=5, last=5) | |
| res = await run_shift(src, db, shift, dst_row, 99, delay=0, settle_pause=0) | |
| assert (res.copied_units, res.replies, res.reply_dropped) == (1, 0, 1) | |
| assert reply_of(dst.msgs[11]) is None | |
| await db.close() | |
| run(go()) | |
| def test_a_source_that_forbids_saving_is_rebuilt_from_its_parts(tmp_path): | |
| async def go(): | |
| src, dst = two_channels() | |
| src.restrict_forwards = True | |
| db, dst_row, shift = await engine_setup(tmp_path, src) | |
| res = await run_shift(src, db, shift, dst_row, 99, delay=0, settle_pause=0) | |
| assert (res.failed, res.forwarded, res.rebuilt) == ([], 0, 6) | |
| new = {i: m for i, m in dst.msgs.items() if i > 10} | |
| assert new[15].grouped_id == new[16].grouped_id and new[13].media.photo.id == 31 | |
| assert reply_of(new[14]) == 12 | |
| await db.close() | |
| run(go()) | |
| def test_a_stopped_shift_continues_without_copying_anything_twice(tmp_path): | |
| async def go(): | |
| src, dst = two_channels() | |
| db, dst_row, shift = await engine_setup(tmp_path, src) | |
| calls = [] | |
| async def prog(res): | |
| calls.append(res.copied_units) | |
| res = await run_shift(src, db, shift, dst_row, 99, delay=0, settle_pause=0, progress=prog, should_stop=lambda: len(calls) >= 2) | |
| assert res.stopped and res.copied_units == 2 and res.final is None | |
| again = await run_shift(src, db, shift, dst_row, 99, delay=0, settle_pause=0) | |
| assert (again.copied_units, again.skipped_done, again.failed) == (4, 2, []) | |
| assert sorted(i for i in dst.msgs if i > 10) == [11, 12, 13, 14, 15, 16, 17] # no duplicates | |
| # the reply of the 3rd copy found the copy made in the first run | |
| assert reply_of(dst.msgs[14]) == 12 | |
| await db.close() | |
| run(go()) | |
| def test_undo_removes_only_the_copies(tmp_path): | |
| async def go(): | |
| src, dst = two_channels() | |
| db, dst_row, shift = await engine_setup(tmp_path, src) | |
| await run_shift(src, db, shift, dst_row, 99, delay=0, settle_pause=0) | |
| res = await delete_shift_copies(src, db, dst_row, shift, delay=0) | |
| assert (res.deleted, res.remaining, res.error) == (7, 0, None) | |
| assert sorted(dst.msgs) == list(range(1, 11)) and sorted(src.msgs) == [1, 2, 3, 4, 5, 6, 7, 8] | |
| assert await db.sent_posts(2) == [] and await db.shift_pairs("s1") == [] | |
| assert (await db.shift_counts("s1"))["new_left"] == 0 | |
| await db.close() | |
| run(go()) | |
| def test_a_post_that_was_shifted_before_is_known(tmp_path): | |
| async def go(): | |
| src, dst = two_channels() | |
| db, dst_row, shift = await engine_setup(tmp_path, src) | |
| await run_shift(src, db, shift, dst_row, 99, delay=0, settle_pause=0) | |
| assert await db.shifted_ids(1, 2, 1, 5) == {1, 2, 3, 5} | |
| assert await db.shifted_ids(1, 3, 1, 5) == set() and await db.shifted_ids(2, 1, 1, 5) == set() | |
| await db.mark_shift_deleted("s1", 2, [11]) | |
| assert await db.shifted_ids(1, 2, 1, 5) == {2, 3, 5} | |
| await db.close() | |
| run(go()) | |
| # ---------------------------------------------------------------------------- as the owner experiences it | |
| async def shift_app(tmp_path, **cfg): | |
| cfg.setdefault("edit_delay", 0.0) | |
| app = App(tmp_path, **cfg) | |
| await app.start() | |
| app.tg.username, app.tg.title = "srcch", "Anime Source" | |
| fill(app.tg, source_posts()) | |
| dst = app.tg.add_channel(FakeChannelClient(old_posts(), username="dstch", cid=2, title="Anime Backup")) | |
| await app.db.save_channel(1, 5, "Anime Source", "srcch", 1) | |
| await app.db.save_channel(2, 6, "Anime Backup", "dstch", 1) | |
| return app, dst | |
| def test_usage_and_bad_commands_are_explained(tmp_path): | |
| async def go(): | |
| app, _ = await shift_app(tmp_path) | |
| await app.text("/shift") | |
| text = norm(app.out.last_text) | |
| assert "Usage: /shift <source> <destination> [from_id] [to_id]" in text and "15 20" in text | |
| for args, part in ( | |
| ("@srcch", "Give the source and the destination"), | |
| ("@srcch @dstch x", "“x” is not a message id"), | |
| ("@srcch @srcch", "same channel"), | |
| ("@srcch @nosuch", "couldn't find a channel"), | |
| ): | |
| await app.text(f"/shift {args}") | |
| assert part in norm(app.out.last_text), (args, norm(app.out.last_text)) | |
| assert not any(isinstance(r, functions.messages.SendMessageRequest) for r in app.tg.requests) | |
| await app.db.close() | |
| run(go()) | |
| def test_the_plan_is_shown_before_anything_is_copied(tmp_path): | |
| async def go(): | |
| app, dst = await shift_app(tmp_path) | |
| await app.text("/shift @srcch @dstch 2 5") | |
| text = norm(app.out.last_text) | |
| assert "Shift plan" in text and "From Anime Source → Anime Backup" in text | |
| assert "Post ids 2-5" in text and "3 posts to copy" in text # 2, 3 and 5 (4 is a service message) | |
| assert "1 service message(s)" in text and "1 post(s) answer another post" in text | |
| assert "1 link(s) in 1 post(s) point at other posts of Anime Source" in text | |
| assert "Nothing is changed in Anime Source" in text | |
| assert dst.requests == [] or not any(isinstance(r, functions.messages.SendMessageRequest) for r in dst.requests) | |
| assert sorted(dst.msgs) == list(range(1, 11)) | |
| await app.db.close() | |
| run(go()) | |
| def test_cancel_and_an_old_button(tmp_path): | |
| async def go(): | |
| app, dst = await shift_app(tmp_path) | |
| await app.text("/shift @srcch @dstch") | |
| start = app.out.callback_data("Start copying") | |
| await app.press(app.out.callback_data("Cancel")) | |
| assert "Cancelled" in norm(app.out.last_text) | |
| await app.press(start) | |
| assert "expired" in norm(app.out.last_text) | |
| assert sorted(dst.msgs) == list(range(1, 11)) | |
| await app.db.close() | |
| run(go()) | |
| def test_the_whole_flow_copy_then_undo(tmp_path): | |
| async def go(): | |
| app, dst = await shift_app(tmp_path) | |
| await app.text("/shift @srcch @dstch") | |
| await app.press(app.out.callback_data("Start copying")) | |
| done = norm(app.out.last_text) | |
| assert "Anime Source → Anime Backup: 7 message(s) copied" in done and "registered in My posts" in done | |
| assert "The source was not changed" in done and "1 link(s) in 1 post(s) now point at the new copies" in done | |
| assert sorted(i for i in dst.msgs if i > 10) == [11, 12, 13, 14, 15, 16, 17] | |
| assert len(await app.db.sent_posts(2)) == 7 | |
| await app.press(app.out.callback_data("Undo")) | |
| assert "Remove the 7 copied message(s) from Anime Backup" in norm(app.out.last_text) | |
| await app.press(app.out.callback_data("Yes, remove")) | |
| assert "7 copied message(s) removed from Anime Backup" in norm(app.out.last_text) | |
| assert sorted(dst.msgs) == list(range(1, 11)) and await app.db.sent_posts(2) == [] | |
| sh = await app.db.open_shift() | |
| assert sh is None # nothing is left open | |
| await app.db.close() | |
| run(go()) | |
| def test_without_arguments_a_finished_shift_can_still_be_undone(tmp_path): | |
| """The result message of a finished shift may be far up the chat: /shift alone offers its Undo again.""" | |
| async def go(): | |
| app, dst = await shift_app(tmp_path) | |
| before = len(app.out.log) | |
| await app.text("/shift") # nothing was shifted yet: only the usage | |
| assert len(app.out.log) - before == 1 and "Usage: /shift" in norm(app.out.last_text) | |
| await app.text("/shift @srcch @dstch") | |
| await app.press(app.out.callback_data("Start copying")) | |
| assert "7 message(s) copied" in norm(app.out.last_text) | |
| await app.text("/shift") | |
| assert "Usage: /shift" in norm(app.out.log[-2][1]) # the usage first, the finished shift right under it | |
| text = norm(app.out.last_text) | |
| assert "Latest finished shift - you can still undo it" in text and "7 message(s) copied" in text | |
| await app.press(app.out.callback_data("Undo")) | |
| await app.press(app.out.callback_data("Yes, remove")) | |
| assert "7 copied message(s) removed from Anime Backup" in norm(app.out.last_text) | |
| assert sorted(dst.msgs) == list(range(1, 11)) | |
| before = len(app.out.log) | |
| await app.text("/shift") # undone: nothing is left to offer | |
| assert len(app.out.log) - before == 1 and "Latest finished shift" not in norm(app.out.last_text) | |
| await app.db.close() | |
| run(go()) | |
| def test_only_a_finished_shift_that_still_has_copies_is_offered(tmp_path): | |
| async def go(): | |
| app, dst = await shift_app(tmp_path) | |
| await app.text("/shift @srcch @dstch") | |
| await app.press(app.out.callback_data("Start copying")) | |
| first = await app.db.last_finished_shift() | |
| assert first is not None | |
| # every copy of it was deleted some other way: nothing to undo, so nothing is offered | |
| await app.db.mark_shift_deleted(first.id, 2, await app.db.shift_pending_new(first.id)) | |
| assert await app.db.last_finished_shift() is None | |
| before = len(app.out.log) | |
| await app.text("/shift") | |
| assert len(app.out.log) - before == 1 and "Latest finished shift" not in norm(app.out.last_text) | |
| # a newer shift is the one offered | |
| await app.text("/shift @srcch @dstch") | |
| await app.press(app.out.callback_data("Start copying")) | |
| second = await app.db.last_finished_shift() | |
| assert second is not None and second.id != first.id | |
| await app.text("/shift") | |
| assert "Latest finished shift" in norm(app.out.last_text) | |
| assert app.out.callback_data("Undo") == f"sfu:{second.id}" | |
| await app.db.close() | |
| run(go()) | |
| def test_a_destination_the_bot_does_not_know_yet_becomes_one_of_its_channels(tmp_path): | |
| async def go(): | |
| app, dst = await shift_app(tmp_path) | |
| await app.db.set_channel_active(2, False) | |
| await app.text("/shift @srcch @dstch 1 1") | |
| await app.press(app.out.callback_data("Start copying")) | |
| assert "1 message(s) copied" in norm(app.out.last_text) | |
| assert [c.id for c in await app.db.list_channels()] == [2, 1] or {c.id for c in await app.db.list_channels()} == {1, 2} | |
| await app.db.close() | |
| run(go()) | |
| def test_the_bot_must_be_allowed_to_post_in_the_destination(tmp_path): | |
| async def go(): | |
| app, dst = await shift_app(tmp_path) | |
| dst.rights = dict(admin=True, post=False, edit=True, delete=True, invite=False, add_admins=False) | |
| await app.text("/shift @srcch @dstch") | |
| assert "needs the “Post messages” admin right in Anime Backup" in norm(app.out.last_text) | |
| await app.db.close() | |
| run(go()) | |
| def test_an_unreadable_source_is_explained(tmp_path): | |
| async def go(): | |
| app, dst = await shift_app(tmp_path) | |
| app.tg.fail["GetMessagesRequest"] = __import__("telethon").errors.ChannelPrivateError(request=None) | |
| await app.text("/shift @srcch @dstch") | |
| text = norm(app.out.last_text) | |
| assert "I can't read Anime Source" in text and "member of the source channel" in text | |
| await app.db.close() | |
| run(go()) | |
| def test_an_unfinished_shift_must_be_dealt_with_first(tmp_path): | |
| async def go(): | |
| app, dst = await shift_app(tmp_path) | |
| await app.db.create_shift("old1", src=SRC, dst_channel_id=2, first_id=1, last_id=8, user_id=1) | |
| await app.db.set_shift_status("old1", "stopped") | |
| await app.text("/shift @srcch @dstch") | |
| text = norm(app.out.last_text) | |
| assert "unfinished shift between these two channels" in text and "stopped" in text | |
| assert "Continue" in " ".join(t for row in app.out.last_buttons() for t in row) | |
| await app.text("/shift") # without arguments it shows the same unfinished shift | |
| assert "Unfinished shift" in norm(app.out.last_text) | |
| # Continue copies what is left; the shift is over afterwards | |
| await app.press(app.out.callback_data("Continue")) | |
| assert "7 message(s) copied" in norm(app.out.last_text) | |
| assert await app.db.open_shift() is None | |
| await app.db.close() | |
| run(go()) | |
| def test_copying_the_same_posts_twice_is_pointed_out(tmp_path): | |
| async def go(): | |
| app, dst = await shift_app(tmp_path) | |
| await app.text("/shift @srcch @dstch 1 2") | |
| await app.press(app.out.callback_data("Start copying")) | |
| await app.text("/shift @srcch @dstch 1 3") | |
| assert "2 of these posts were already shifted into Anime Backup before" in norm(app.out.last_text) | |
| await app.db.close() | |
| run(go()) | |
| def test_shift_is_for_the_owner_only(tmp_path): | |
| async def go(): | |
| app, dst = await shift_app(tmp_path, admins=frozenset({7})) | |
| app.uid = 7 | |
| await app.text("/shift @srcch @dstch") | |
| assert "Shift plan" not in norm(app.out.last_text or "") | |
| assert sorted(dst.msgs) == list(range(1, 11)) | |
| await app.db.close() | |
| run(go()) | |