Coverage for src/tests/test_discussions.py: 100%

1import grpc 

2import pytest 

3 

4from couchers import errors 

5from couchers.db import session_scope 

6from couchers.utils import now, to_aware_datetime 

7from proto import discussions_pb2, notifications_pb2, threads_pb2 

8from tests.test_communities import create_community, create_group 

9from tests.test_fixtures import ( # noqa 

10 db, 

11 discussions_session, 

12 generate_user, 

13 notifications_session, 

14 process_jobs, 

15 push_collector, 

16 testconfig, 

17 threads_session, 

18) 

19 

20 

21@pytest.fixture(autouse=True) 

22def _(testconfig): 

23 pass 

24 

25 

26def test_create_discussion_errors(db): 

27 user, token = generate_user() 

28 with discussions_session(token) as api: 

29 with pytest.raises(grpc.RpcError) as e: 

30 api.CreateDiscussion( 

31 discussions_pb2.CreateDiscussionReq( 

32 title=None, 

33 content="dummy content", 

34 ) 

35 ) 

36 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT 

37 assert e.value.details() == errors.MISSING_DISCUSSION_TITLE 

38 

39 with discussions_session(token) as api: 

40 with pytest.raises(grpc.RpcError) as e: 

41 api.CreateDiscussion( 

42 discussions_pb2.CreateDiscussionReq( 

43 title="dummy title", 

44 content=None, 

45 ) 

46 ) 

47 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT 

48 assert e.value.details() == errors.MISSING_DISCUSSION_CONTENT 

49 

50 

51def test_create_and_get_discussion(db, push_collector): 

52 generate_user() 

53 user, token = generate_user() 

54 user2, token2 = generate_user() 

55 generate_user() 

56 generate_user() 

57 

58 with notifications_session(token2) as notifications: 

59 notifications.SetNotificationSettings( 

60 notifications_pb2.SetNotificationSettingsReq( 

61 preferences=[ 

62 notifications_pb2.SingleNotificationPreference( 

63 topic="discussion", 

64 action="create", 

65 delivery_method="push", 

66 enabled=True, 

67 ) 

68 ], 

69 ) 

70 ) 

71 

72 with session_scope() as session: 

73 community = create_community(session, 0, 1, "Testing Community", [user2], [], None) 

74 group_id = create_group(session, "Testing Group", [user2], [], community).id 

75 community_id = community.id 

76 user2_id = user2.id 

77 

78 with discussions_session(token) as api: 

79 time_before_create = now() 

80 res = api.CreateDiscussion( 

81 discussions_pb2.CreateDiscussionReq( 

82 title="dummy title", 

83 content="dummy content", 

84 owner_community_id=community_id, 

85 ) 

86 ) 

87 time_after_create = now() 

88 

89 assert res.title == "dummy title" 

90 assert res.content == "dummy content" 

91 assert res.slug == "dummy-title" 

92 assert time_before_create <= to_aware_datetime(res.created) <= time_after_create 

93 assert res.creator_user_id == user.id 

94 assert res.owner_community_id == community_id 

95 

96 discussion_id = res.discussion_id 

97 

98 process_jobs() 

99 

100 push_collector.assert_user_has_single_matching( 

101 user2_id, 

102 title="dummy title", 

103 body=f"{user.name} created a discussion in Testing Community: dummy title\n\ndummy content", 

104 ) 

105 

106 with discussions_session(token) as api: 

107 res = api.GetDiscussion( 

108 discussions_pb2.GetDiscussionReq( 

109 discussion_id=discussion_id, 

110 ) 

111 ) 

112 

113 assert res.title == "dummy title" 

114 assert res.content == "dummy content" 

115 assert res.slug == "dummy-title" 

116 assert time_before_create <= to_aware_datetime(res.created) <= time_after_create 

117 assert res.creator_user_id == user.id 

118 assert res.owner_community_id == community_id 

119 

120 with discussions_session(token) as api: 

121 time_before_create = now() 

122 res = api.CreateDiscussion( 

123 discussions_pb2.CreateDiscussionReq( 

124 title="dummy title", 

125 content="dummy content", 

126 owner_group_id=group_id, 

127 ) 

128 ) 

129 time_after_create = now() 

130 

131 assert res.title == "dummy title" 

132 assert res.content == "dummy content" 

133 assert res.slug == "dummy-title" 

134 assert time_before_create <= to_aware_datetime(res.created) <= time_after_create 

135 assert res.creator_user_id == user.id 

136 assert res.owner_group_id == group_id 

137 

138 discussion_id = res.discussion_id 

139 

140 with discussions_session(token) as api: 

141 res = api.GetDiscussion( 

142 discussions_pb2.GetDiscussionReq( 

143 discussion_id=discussion_id, 

144 ) 

145 ) 

146 

147 assert res.title == "dummy title" 

148 assert res.content == "dummy content" 

149 assert res.slug == "dummy-title" 

150 assert time_before_create <= to_aware_datetime(res.created) <= time_after_create 

151 assert res.creator_user_id == user.id 

152 assert res.owner_group_id == group_id 

153 

154 

155def test_discussion_notifications_regression(db, push_collector): 

156 generate_user() 

157 user, token = generate_user() 

158 user2, token2 = generate_user() 

159 generate_user() 

160 generate_user() 

161 

162 with session_scope() as session: 

163 community = create_community(session, 0, 1, "Testing Community", [user2], [], None) 

164 group_id = create_group(session, "Testing Group", [user2], [], community).id 

165 community_id = community.id 

166 user2_id = user2.id 

167 

168 with discussions_session(token) as api: 

169 time_before_create = now() 

170 res = api.CreateDiscussion( 

171 discussions_pb2.CreateDiscussionReq( 

172 title="dummy title", 

173 content="dummy content", 

174 owner_community_id=community_id, 

175 ) 

176 ) 

177 time_after_create = now() 

178 

179 assert res.title == "dummy title" 

180 assert res.content == "dummy content" 

181 assert res.slug == "dummy-title" 

182 assert time_before_create <= to_aware_datetime(res.created) <= time_after_create 

183 assert res.creator_user_id == user.id 

184 assert res.owner_community_id == community_id 

185 

186 discussion_id = res.discussion_id 

187 thread_id = res.thread.thread_id 

188 

189 with threads_session(token2) as api: 

190 reply_id = api.PostReply(threads_pb2.PostReplyReq(thread_id=thread_id, content="hi")).thread_id 

191 

192 with threads_session(token) as api: 

193 api.PostReply(threads_pb2.PostReplyReq(thread_id=reply_id, content="what a silly comment")) 

194 

195 process_jobs() 

196 

197 push_collector.assert_user_has_single_matching( 

198 user2_id, 

199 title="dummy title", 

200 )