karaolidis/fbchat
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Split Plan into Plan/PlanData, and add Plan.session

Browse Source
This commit is contained in:
Mads Marquart
2020-01-09 15:05:42 +01:00
parent e039e88f80
commit 483fdf43dc
8 changed files with 64 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
fbchat/__init__.py
View File

@@ -31,7 +31,7 @@ from ._quick_reply import (
QuickReplyEmail, QuickReplyEmail,
) )
from ._poll import Poll, PollOption from ._poll import Poll, PollOption
from ._plan import GuestStatus, Plan from ._plan import GuestStatus, Plan, PlanData
from ._client import Client from ._client import Client

14
fbchat/_client.py
View File

@@ -24,7 +24,7 @@ from ._quick_reply import (
QuickReplyEmail, QuickReplyEmail,
) )
from ._poll import Poll, PollOption from ._poll import Poll, PollOption
from ._plan import ACONTEXT, Plan from ._plan import ACONTEXT, PlanData
class Client: class Client:
@@ -646,7 +646,7 @@ class Client:
""" """
data = {"event_reminder_id": plan_id} data = {"event_reminder_id": plan_id}
j = self._payload_post("/ajax/eventreminder", data) j = self._payload_post("/ajax/eventreminder", data)
return Plan._from_fetch(j) return PlanData._from_fetch(self.session, j)
def _get_private_data(self): def _get_private_data(self):
(j,) = self.graphql_requests(_graphql.from_doc_id("1868889766468115", {})) (j,) = self.graphql_requests(_graphql.from_doc_id("1868889766468115", {}))
@@ -1284,7 +1284,7 @@ class Client:
elif delta_type == "lightweight_event_create": elif delta_type == "lightweight_event_create":
self.on_plan_created( self.on_plan_created(
mid=mid, mid=mid,
plan=Plan._from_pull(delta["untypedData"]), plan=PlanData._from_pull(self.session, delta["untypedData"]),
author_id=author_id, author_id=author_id,
thread=get_thread(metadata), thread=get_thread(metadata),
at=at, at=at,
@@ -1296,7 +1296,7 @@ class Client:
elif delta_type == "lightweight_event_notify": elif delta_type == "lightweight_event_notify":
self.on_plan_ended( self.on_plan_ended(
mid=mid, mid=mid,
plan=Plan._from_pull(delta["untypedData"]), plan=PlanData._from_pull(self.session, delta["untypedData"]),
thread=get_thread(metadata), thread=get_thread(metadata),
at=at, at=at,
metadata=metadata, metadata=metadata,
@@ -1307,7 +1307,7 @@ class Client:
elif delta_type == "lightweight_event_update": elif delta_type == "lightweight_event_update":
self.on_plan_edited( self.on_plan_edited(
mid=mid, mid=mid,
plan=Plan._from_pull(delta["untypedData"]), plan=PlanData._from_pull(self.session, delta["untypedData"]),
author_id=author_id, author_id=author_id,
thread=get_thread(metadata), thread=get_thread(metadata),
at=at, at=at,
@@ -1319,7 +1319,7 @@ class Client:
elif delta_type == "lightweight_event_delete": elif delta_type == "lightweight_event_delete":
self.on_plan_deleted( self.on_plan_deleted(
mid=mid, mid=mid,
plan=Plan._from_pull(delta["untypedData"]), plan=PlanData._from_pull(self.session, delta["untypedData"]),
author_id=author_id, author_id=author_id,
thread=get_thread(metadata), thread=get_thread(metadata),
at=at, at=at,
@@ -1332,7 +1332,7 @@ class Client:
take_part = delta["untypedData"]["guest_status"] == "GOING" take_part = delta["untypedData"]["guest_status"] == "GOING"
self.on_plan_participation( self.on_plan_participation(
mid=mid, mid=mid,
plan=Plan._from_pull(delta["untypedData"]), plan=PlanData._from_pull(self.session, delta["untypedData"]),
take_part=take_part, take_part=take_part,
author_id=author_id, author_id=author_id,
thread=get_thread(metadata), thread=get_thread(metadata),

4
fbchat/_group.py
View File

@@ -175,7 +175,9 @@ class GroupData(Group):
) )
plan = None plan = None
if data.get("event_reminders") and data["event_reminders"].get("nodes"): if data.get("event_reminders") and data["event_reminders"].get("nodes"):
plan = _plan.Plan._from_graphql(data["event_reminders"]["nodes"][0]) plan = _plan.PlanData._from_graphql(
session, data["event_reminders"]["nodes"][0]
)
return cls( return cls(
session=session, session=session,

4
fbchat/_page.py
View File

@@ -52,7 +52,9 @@ class PageData(Page):
data["city"] = {} data["city"] = {}
plan = None plan = None
if data.get("event_reminders") and data["event_reminders"].get("nodes"): if data.get("event_reminders") and data["event_reminders"].get("nodes"):
plan = _plan.Plan._from_graphql(data["event_reminders"]["nodes"][0]) plan = _plan.PlanData._from_graphql(
session, data["event_reminders"]["nodes"][0]
)
return cls( return cls(
session=session, session=session,

25
fbchat/_plan.py
View File

@@ -1,7 +1,7 @@
import attr import attr
import json import json
from ._core import attrs_default, Enum from ._core import attrs_default, Enum
from . import _util from . import _util, _session
class GuestStatus(Enum): class GuestStatus(Enum):
@@ -19,14 +19,22 @@ ACONTEXT = {
@attrs_default @attrs_default
class Plan: class Plan:
"""Represents a plan.""" """Base model for plans."""
#: The session to use when making requests.
session = attr.ib(type=_session.Session)
#: The plan's unique identifier.
id = attr.ib(converter=str)
@attrs_default
class PlanData(Plan):
"""Represents data about a plan."""
#: Plan time (datetime), only precise down to the minute #: Plan time (datetime), only precise down to the minute
time = attr.ib() time = attr.ib()
#: Plan title #: Plan title
title = attr.ib() title = attr.ib()
#: ID of the plan
id = attr.ib(None)
#: Plan location name #: Plan location name
location = attr.ib(None, converter=lambda x: x or "") location = attr.ib(None, converter=lambda x: x or "")
#: Plan location ID #: Plan location ID
@@ -64,8 +72,9 @@ class Plan:
] ]
@classmethod @classmethod
def _from_pull(cls, data): def _from_pull(cls, session, data):
return cls( return cls(
session=session,
id=data.get("event_id"), id=data.get("event_id"),
time=_util.seconds_to_datetime(int(data.get("event_time"))), time=_util.seconds_to_datetime(int(data.get("event_time"))),
title=data.get("event_title"), title=data.get("event_title"),
@@ -79,8 +88,9 @@ class Plan:
) )
@classmethod @classmethod
def _from_fetch(cls, data): def _from_fetch(cls, session, data):
return cls( return cls(
session=session,
id=data.get("oid"), id=data.get("oid"),
time=_util.seconds_to_datetime(data.get("event_time")), time=_util.seconds_to_datetime(data.get("event_time")),
title=data.get("title"), title=data.get("title"),
@@ -91,8 +101,9 @@ class Plan:
) )
@classmethod @classmethod
def _from_graphql(cls, data): def _from_graphql(cls, session, data):
return cls( return cls(
session=session,
id=data.get("id"), id=data.get("id"),
time=_util.seconds_to_datetime(data.get("time")), time=_util.seconds_to_datetime(data.get("time")),
title=data.get("event_title"), title=data.get("event_title"),

8
fbchat/_user.py
View File

@@ -118,7 +118,9 @@ class UserData(User):
c_info = cls._parse_customization_info(data) c_info = cls._parse_customization_info(data)
plan = None plan = None
if data.get("event_reminders") and data["event_reminders"].get("nodes"): if data.get("event_reminders") and data["event_reminders"].get("nodes"):
plan = _plan.Plan._from_graphql(data["event_reminders"]["nodes"][0]) plan = _plan.PlanData._from_graphql(
session, data["event_reminders"]["nodes"][0]
)
return cls( return cls(
session=session, session=session,
@@ -166,7 +168,9 @@ class UserData(User):
plan = None plan = None
if data.get("event_reminders") and data["event_reminders"].get("nodes"): if data.get("event_reminders") and data["event_reminders"].get("nodes"):
plan = _plan.Plan._from_graphql(data["event_reminders"]["nodes"][0]) plan = _plan.PlanData._from_graphql(
session, data["event_reminders"]["nodes"][0]
)
return cls( return cls(
session=session, session=session,

29
tests/test_plan.py
View File

@@ -1,9 +1,11 @@
import datetime import datetime
from fbchat._plan import GuestStatus, Plan from fbchat._plan import GuestStatus, PlanData
def test_plan_properties(): def test_plan_properties(session):
plan = Plan( plan = PlanData(
session=session,
id="1234567890",
time=..., time=...,
title=..., title=...,
guests={ guests={
@@ -18,7 +20,7 @@ def test_plan_properties():
assert plan.declined == ["4567"] assert plan.declined == ["4567"]
def test_plan_from_pull(): def test_plan_from_pull(session):
data = { data = {
"event_timezone": "", "event_timezone": "",
"event_creator_id": "1234", "event_creator_id": "1234",
@@ -35,7 +37,8 @@ def test_plan_from_pull():
'{"guest_list_state":"GOING","node":{"id":"4567"}}]' '{"guest_list_state":"GOING","node":{"id":"4567"}}]'
), ),
} }
assert Plan( assert PlanData(
session=session,
id="1111", id="1111",
time=datetime.datetime(2017, 7, 14, 2, 40, tzinfo=datetime.timezone.utc), time=datetime.datetime(2017, 7, 14, 2, 40, tzinfo=datetime.timezone.utc),
title="abc", title="abc",
@@ -46,10 +49,10 @@ def test_plan_from_pull():
"3456": GuestStatus.DECLINED, "3456": GuestStatus.DECLINED,
"4567": GuestStatus.GOING, "4567": GuestStatus.GOING,
}, },
) == Plan._from_pull(data) ) == PlanData._from_pull(session, data)
def test_plan_from_fetch(): def test_plan_from_fetch(session):
data = { data = {
"message_thread_id": 123456789, "message_thread_id": 123456789,
"event_time": 1500000000, "event_time": 1500000000,
@@ -92,7 +95,8 @@ def test_plan_from_fetch():
"4567": "GOING", "4567": "GOING",
}, },
} }
assert Plan( assert PlanData(
session=session,
id=1111, id=1111,
time=datetime.datetime(2017, 7, 14, 2, 40, tzinfo=datetime.timezone.utc), time=datetime.datetime(2017, 7, 14, 2, 40, tzinfo=datetime.timezone.utc),
title="abc", title="abc",
@@ -105,10 +109,10 @@ def test_plan_from_fetch():
"3456": GuestStatus.DECLINED, "3456": GuestStatus.DECLINED,
"4567": GuestStatus.GOING, "4567": GuestStatus.GOING,
}, },
) == Plan._from_fetch(data) ) == PlanData._from_fetch(session, data)
def test_plan_from_graphql(): def test_plan_from_graphql(session):
data = { data = {
"id": "1111", "id": "1111",
"lightweight_event_creator": {"id": "1234"}, "lightweight_event_creator": {"id": "1234"},
@@ -134,7 +138,8 @@ def test_plan_from_graphql():
] ]
}, },
} }
assert Plan( assert PlanData(
session=session,
time=datetime.datetime(2017, 7, 14, 2, 40, tzinfo=datetime.timezone.utc), time=datetime.datetime(2017, 7, 14, 2, 40, tzinfo=datetime.timezone.utc),
title="abc", title="abc",
location="", location="",
@@ -147,4 +152,4 @@ def test_plan_from_graphql():
"3456": GuestStatus.DECLINED, "3456": GuestStatus.DECLINED,
"4567": GuestStatus.GOING, "4567": GuestStatus.GOING,
}, },
) == Plan._from_graphql(data) ) == PlanData._from_graphql(session, data)

18
tests/test_plans.py
View File

@@ -1,6 +1,6 @@
import pytest import pytest
from fbchat import Plan, FBchatFacebookError from fbchat import PlanData, FBchatFacebookError
from utils import random_hex, subset from utils import random_hex, subset
from time import time from time import time
@@ -10,12 +10,12 @@ pytestmark = pytest.mark.online
@pytest.fixture( @pytest.fixture(
scope="module", scope="module",
params=[ params=[
Plan(time=int(time()) + 100, title=random_hex()), # PlanData(time=int(time()) + 100, title=random_hex()),
pytest.param( # pytest.param(
Plan(time=int(time()), title=random_hex()), # PlanData(time=int(time()), title=random_hex()),
marks=[pytest.mark.xfail(raises=FBchatFacebookError)], # marks=[pytest.mark.xfail(raises=FBchatFacebookError)],
), # ),
pytest.param(Plan(time=0, title=None), marks=[pytest.mark.xfail()]), # pytest.param(PlanData(time=0, title=None), marks=[pytest.mark.xfail()]),
], ],
) )
def plan_data(request, client, user, thread, catch_event, compare): def plan_data(request, client, user, thread, catch_event, compare):
@@ -73,7 +73,7 @@ def test_change_plan_participation(
@pytest.mark.trylast @pytest.mark.trylast
def test_edit_plan(client, thread, catch_event, compare, plan_data): def test_edit_plan(client, thread, catch_event, compare, plan_data):
event, plan = plan_data event, plan = plan_data
new_plan = Plan(plan.time + 100, random_hex()) new_plan = PlanData(plan.time + 100, random_hex())
with catch_event("on_plan_edited") as x: with catch_event("on_plan_edited") as x:
client.edit_plan(plan, new_plan) client.edit_plan(plan, new_plan)
assert compare(x) assert compare(x)
@@ -89,7 +89,7 @@ def test_edit_plan(client, thread, catch_event, compare, plan_data):
@pytest.mark.skip @pytest.mark.skip
def test_on_plan_ended(client, thread, catch_event, compare): def test_on_plan_ended(client, thread, catch_event, compare):
with catch_event("on_plan_ended") as x: with catch_event("on_plan_ended") as x:
client.create_plan(Plan(int(time()) + 120, "Wait for ending")) client.create_plan(PlanData(int(time()) + 120, "Wait for ending"))
x.wait(180) x.wait(180)
assert subset( assert subset(
x.res, x.res,