本文整理汇总了Python中zerver.lib.actions.create_stream_if_needed函数的典型用法代码示例。如果您正苦于以下问题:Python create_stream_if_needed函数的具体用法?Python create_stream_if_needed怎么用?Python create_stream_if_needed使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了create_stream_if_needed函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: handle
def handle(self, *args: Any, **options: str) -> None:
realm = self.get_realm(options)
assert realm is not None # Should be ensured by parser
encoding = sys.getfilesystemencoding()
stream_name = options['stream_name']
create_stream_if_needed(realm, force_text(stream_name, encoding))
开发者ID:brockwhittaker,项目名称:zulip,代码行数:7,代码来源:create_stream.py
示例2: test_new_stream_link
def test_new_stream_link(self, mock_django_timezone: mock.MagicMock) -> None:
cutoff = datetime.datetime(year=2017, month=11, day=1)
mock_django_timezone.return_value = datetime.datetime(year=2017, month=11, day=5)
cordelia = self.example_user('cordelia')
create_stream_if_needed(cordelia.realm, 'New stream')
new_stream = gather_new_streams(cordelia, cutoff)[1]
expected_html = "<a href='http://zulip.testserver/#narrow/stream/New.20stream'>New stream</a>"
self.assertIn(expected_html, new_stream['html'])
开发者ID:joydeep1701,项目名称:zulip,代码行数:8,代码来源:test_email_mirror.py
示例3: handle
def handle(self, *args, **options):
# type: (*Any, **str) -> None
string_id = options['realm']
encoding = sys.getfilesystemencoding()
stream_name = options['stream_name']
realm = get_realm(force_text(string_id, encoding))
if realm is None:
print("Unknown string_id %s" % (string_id,))
exit(1)
create_stream_if_needed(realm, force_text(stream_name, encoding))
开发者ID:JamesLinus,项目名称:zulip,代码行数:12,代码来源:create_stream.py
示例4: test_register
def test_register(self):
realm = get_realm("zulip.com")
streams = ["stream_%s" % i for i in range(40)]
for stream in streams:
create_stream_if_needed(realm, stream)
set_default_streams(realm, streams)
with queries_captured() as queries:
self.register("test", "test")
# Ensure the number of queries we make is not O(streams)
self.assert_length(queries, 74)
user_profile = get_user_profile_by_email('[email protected]')
self.assertEqual(get_session_dict_user(self.client.session), user_profile.id)
开发者ID:anindya,项目名称:zulip,代码行数:13,代码来源:test_signup.py
示例5: handle
def handle(self, **options):
# type: (**Any) -> None
if options["domain"] is None or options["streams"] is None or \
(options["users"] is None and options["all_users"] is None):
self.print_help("python manage.py", "add_users_to_streams")
exit(1)
stream_names = set([stream.strip() for stream in options["streams"].split(",")])
realm = get_realm(options["domain"])
if options["all_users"]:
user_profiles = UserProfile.objects.filter(realm=realm)
else:
emails = set([email.strip() for email in options["users"].split(",")])
user_profiles = []
for email in emails:
user_profiles.append(get_user_profile_by_email(email))
for stream_name in set(stream_names):
for user_profile in user_profiles:
stream, _ = create_stream_if_needed(user_profile.realm, stream_name)
did_subscribe = do_add_subscription(user_profile, stream)
print("%s %s to %s" % (
"Subscribed" if did_subscribe else "Already subscribed",
user_profile.email, stream_name))
开发者ID:150vb,项目名称:zulip,代码行数:25,代码来源:add_users_to_streams.py
示例6: test_rename_stream
def test_rename_stream(self):
realm = get_realm('zulip.com')
stream, _ = create_stream_if_needed(realm, 'old_name')
new_name = u'stream with a brand new name'
self.subscribe_to_stream(self.user_profile.email, stream.name)
action = lambda: do_rename_stream(realm, stream.name, new_name)
events = self.do_test(action)
schema_checker = check_dict([
('type', equals('stream')),
('op', equals('update')),
('property', equals('email_address')),
('value', check_string),
('name', equals('old_name')),
])
error = schema_checker('events[0]', events[0])
self.assert_on_error(error)
schema_checker = check_dict([
('type', equals('stream')),
('op', equals('update')),
('property', equals('name')),
('value', equals(new_name)),
('name', equals('old_name')),
])
error = schema_checker('events[1]', events[1])
self.assert_on_error(error)
开发者ID:seanly,项目名称:zulip,代码行数:28,代码来源:test_events.py
示例7: subscribe_to_stream
def subscribe_to_stream(self, email, stream_name, realm=None):
realm = get_realm(resolve_email_to_domain(email))
stream = get_stream(stream_name, realm)
if stream is None:
stream, _ = create_stream_if_needed(realm, stream_name)
user_profile = get_user_profile_by_email(email)
do_add_subscription(user_profile, stream, no_log=True)
开发者ID:RomiPierre,项目名称:zulip,代码行数:7,代码来源:test_helpers.py
示例8: test_invite_user_signup_initial_history
def test_invite_user_signup_initial_history(self):
# type: () -> None
"""
Test that a new user invited to a stream receives some initial
history but only from public streams.
"""
self.login("[email protected]")
user_profile = get_user_profile_by_email("[email protected]")
private_stream_name = "Secret"
(stream, _) = create_stream_if_needed(user_profile.realm, private_stream_name, invite_only=True)
do_add_subscription(user_profile, stream)
public_msg_id = self.send_message("[email protected]", "Denmark", Recipient.STREAM,
"Public topic", "Public message")
secret_msg_id = self.send_message("[email protected]", private_stream_name, Recipient.STREAM,
"Secret topic", "Secret message")
invitee = "[email protected]"
self.assert_json_success(self.invite(invitee, [private_stream_name, "Denmark"]))
self.assertTrue(find_key_by_email(invitee))
self.submit_reg_form_for_user("alice-test", "password")
invitee_profile = get_user_profile_by_email(invitee)
invitee_msg_ids = [um.message_id for um in
UserMessage.objects.filter(user_profile=invitee_profile)]
self.assertTrue(public_msg_id in invitee_msg_ids)
self.assertFalse(secret_msg_id in invitee_msg_ids)
开发者ID:tobby2002,项目名称:zulip,代码行数:25,代码来源:test_signup.py
示例9: handle
def handle(self, **options):
# type: (**Any) -> None
if (
options["string_id"] is None
or options["streams"] is None
or (options["users"] is None and options["all_users"] is None)
):
self.print_help("./manage.py", "add_users_to_streams")
exit(1)
stream_names = set([stream.strip() for stream in options["streams"].split(",")])
realm = get_realm_by_string_id(options["string_id"])
if options["all_users"]:
user_profiles = UserProfile.objects.filter(realm=realm)
else:
emails = set([email.strip() for email in options["users"].split(",")])
user_profiles = []
for email in emails:
user_profiles.append(get_user_profile_by_email(email))
for stream_name in set(stream_names):
for user_profile in user_profiles:
stream, _ = create_stream_if_needed(user_profile.realm, stream_name)
_ignore, already_subscribed = bulk_add_subscriptions([stream], [user_profile])
was_there_already = user_profile.id in {tup[0].id for tup in already_subscribed}
print(
"%s %s to %s"
% ("Already subscribed" if was_there_already else "Subscribed", user_profile.email, stream_name)
)
开发者ID:zulip,项目名称:zulip,代码行数:30,代码来源:add_users_to_streams.py
示例10: handle
def handle(self, *args, **options):
# type: (*Any, **Any) -> None
realm = self.get_realm(options)
assert realm is not None # Should be ensured by parser
streams = []
stream_names = set([stream.strip() for stream in options["streams"].split(",")])
for stream_name in set(stream_names):
stream, _ = create_stream_if_needed(realm, stream_name)
streams.append(stream)
try:
default_stream_group = DefaultStreamGroup.objects.get(
name=options["name"], realm=realm, description=options["description"])
except DefaultStreamGroup.DoesNotExist:
default_stream_group = DefaultStreamGroup.objects.create(
name=options["name"], realm=realm, description=options["description"])
default_stream_group.streams = streams
default_stream_group.save()
default_stream_groups = DefaultStreamGroup.objects.all()
for default_stream_group in default_stream_groups:
print(default_stream_group.name)
print(default_stream_group.description)
for stream in default_stream_group.streams.all():
print(stream.name)
print("")
开发者ID:joydeep1701,项目名称:zulip,代码行数:27,代码来源:create_default_stream_groups.py
示例11: subscribe
def subscribe(self, user_profile: UserProfile, stream_name: str) -> Stream:
try:
stream = get_stream(stream_name, user_profile.realm)
from_stream_creation = False
except Stream.DoesNotExist:
stream, from_stream_creation = create_stream_if_needed(user_profile.realm, stream_name)
bulk_add_subscriptions([stream], [user_profile], from_stream_creation=from_stream_creation)
return stream
开发者ID:BakerWang,项目名称:zulip,代码行数:8,代码来源:test_classes.py
示例12: subscribe_to_stream
def subscribe_to_stream(self, email, stream_name, realm=None):
# type: (text_type, text_type, Optional[Realm]) -> None
if realm is None:
realm = get_realm_by_email_domain(email)
stream = get_stream(stream_name, realm)
if stream is None:
stream, _ = create_stream_if_needed(realm, stream_name)
user_profile = get_user_profile_by_email(email)
bulk_add_subscriptions([stream], [user_profile])
开发者ID:shekhirin,项目名称:zulip,代码行数:9,代码来源:test_classes.py
示例13: list_to_streams
def list_to_streams(streams_raw, user_profile, autocreate=False, invite_only=False):
# type: (Iterable[text_type], UserProfile, Optional[bool], Optional[bool]) -> Tuple[List[Stream], List[Stream]]
"""Converts plaintext stream names to a list of Streams, validating input in the process
For each stream name, we validate it to ensure it meets our
requirements for a proper stream name: that is, that it is shorter
than Stream.MAX_NAME_LENGTH characters and passes
valid_stream_name.
This function in autocreate mode should be atomic: either an exception will be raised
during a precheck, or all the streams specified will have been created if applicable.
@param streams_raw The list of stream names to process
@param user_profile The user for whom we are retreiving the streams
@param autocreate Whether we should create streams if they don't already exist
@param invite_only Whether newly created streams should have the invite_only bit set
"""
existing_streams = []
created_streams = []
# Validate all streams, getting extant ones, then get-or-creating the rest.
stream_set = set(stream_name.strip() for stream_name in streams_raw)
rejects = []
for stream_name in stream_set:
if len(stream_name) > Stream.MAX_NAME_LENGTH:
raise JsonableError(_("Stream name (%s) too long.") % (stream_name,))
if not valid_stream_name(stream_name):
raise JsonableError(_("Invalid stream name (%s).") % (stream_name,))
existing_stream_map = bulk_get_streams(user_profile.realm, stream_set)
for stream_name in stream_set:
stream = existing_stream_map.get(stream_name.lower())
if stream is None:
rejects.append(stream_name)
else:
existing_streams.append(stream)
if rejects:
if not user_profile.can_create_streams():
raise JsonableError(_('User cannot create streams.'))
elif not autocreate:
raise JsonableError(_("Stream(s) (%s) do not exist") % ", ".join(rejects))
for stream_name in rejects:
stream, created = create_stream_if_needed(user_profile.realm,
stream_name,
invite_only=invite_only)
if created:
created_streams.append(stream)
else:
# We already checked for existing streams above; this
# next line is present to handle races where a stream
# was created while this function was executing.
existing_streams.append(stream)
return existing_streams, created_streams
开发者ID:Kingedgar,项目名称:zulip,代码行数:55,代码来源:streams.py
示例14: subscribe_to_stream
def subscribe_to_stream(self, email, stream_name, realm=None):
# type: (Text, Text, Optional[Realm]) -> Stream
if realm is None:
realm = get_realm_by_email_domain(email)
try:
stream = get_stream(stream_name, realm)
from_creation = False
except Stream.DoesNotExist:
stream, from_creation = create_stream_if_needed(realm, stream_name)
user_profile = get_user_profile_by_email(email)
bulk_add_subscriptions([stream], [user_profile], from_creation=from_creation)
return stream
开发者ID:christi3k,项目名称:zulip,代码行数:12,代码来源:test_classes.py
示例15: handle
def handle(self, *args: Any, **options: Any) -> None:
realm = self.get_realm(options)
assert realm is not None # Should be ensured by parser
streams = [] # type: List[Stream]
if options["streams"]:
stream_names = set([stream.strip() for stream in options["streams"].split(",")])
for stream_name in set(stream_names):
stream, _ = create_stream_if_needed(realm, stream_name)
streams.append(stream)
referred_by = self.get_user(options['referred_by'], realm)
invite_link = do_create_multiuse_invite_link(referred_by, streams)
print("You can use %s to invite as many number of people to the organization." % (invite_link,))
开发者ID:gnprice,项目名称:zulip,代码行数:14,代码来源:generate_multiuse_invite_link.py
示例16: handle
def handle(self, **options: Any) -> None:
realm = self.get_realm(options)
assert realm is not None # Should be ensured by parser
user_profiles = self.get_users(options, realm)
stream_names = set([stream.strip() for stream in options["streams"].split(",")])
for stream_name in set(stream_names):
for user_profile in user_profiles:
stream, _ = create_stream_if_needed(realm, stream_name)
_ignore, already_subscribed = bulk_add_subscriptions([stream], [user_profile])
was_there_already = user_profile.id in {tup[0].id for tup in already_subscribed}
print("%s %s to %s" % (
"Already subscribed" if was_there_already else "Subscribed",
user_profile.email, stream_name))
开发者ID:brockwhittaker,项目名称:zulip,代码行数:15,代码来源:add_users_to_streams.py
示例17: test_invite_with_non_ascii_streams
def test_invite_with_non_ascii_streams(self):
"""
Inviting someone to streams with non-ASCII characters succeeds.
"""
self.login("[email protected]")
invitee = "[email protected]"
stream_name = u"hümbüǵ"
realm = get_realm("zulip.com")
stream, _ = create_stream_if_needed(realm, stream_name)
# Make sure we're subscribed before inviting someone.
do_add_subscription(
get_user_profile_by_email("[email protected]"),
stream, no_log=True)
self.assert_json_success(self.invite(invitee, [stream_name]))
开发者ID:anindya,项目名称:zulip,代码行数:17,代码来源:test_signup.py
示例18: handle
def handle(self, *args: Any, **options: Any) -> None:
realm = self.get_realm(options)
assert realm is not None # Should be ensured by parser
streams = []
if options["streams"]:
stream_names = set([stream.strip() for stream in options["streams"].split(",")])
for stream_name in set(stream_names):
stream, _ = create_stream_if_needed(realm, stream_name)
streams.append(stream)
referred_by = self.get_user(options['referred_by'], realm)
invite = MultiuseInvite.objects.create(realm=realm, referred_by=referred_by)
if streams:
invite.streams = streams
invite.save()
invite_link = create_confirmation_link(invite, realm.host, Confirmation.MULTIUSE_INVITE)
print("You can use %s to invite as many number of people to the organization." % (invite_link,))
开发者ID:joydeep1701,项目名称:zulip,代码行数:21,代码来源:generate_multiuse_invite_link.py
示例19: subscribe_to_stream
def subscribe_to_stream(self, email, stream_name, realm=None):
realm = Realm.objects.get(domain=resolve_email_to_domain(email))
stream, _ = create_stream_if_needed(realm, stream_name)
user_profile = get_user_profile_by_email(email)
do_add_subscription(user_profile, stream, no_log=True)
开发者ID:Gabriel0402,项目名称:zulip,代码行数:5,代码来源:test_helpers.py
示例20: add_message_formatting_conversation
def add_message_formatting_conversation(self) -> None:
realm = get_realm('zulip')
stream, _ = create_stream_if_needed(realm, 'zulip features')
UserProfile.objects.filter(email__contains='stage').delete()
starr = do_create_user('[email protected]', 'password', realm, 'Ada Starr', '')
self.set_avatar(starr, 'static/images/features/starr.png')
fisher = do_create_user('[email protected]', 'password', realm, 'Bel Fisher', '')
self.set_avatar(fisher, 'static/images/features/fisher.png')
twitter_bot = do_create_user('[email protected]', 'password', realm, 'Twitter Bot', '',
bot_type=UserProfile.DEFAULT_BOT)
self.set_avatar(twitter_bot, 'static/images/features/twitter.png')
bulk_add_subscriptions([stream], list(UserProfile.objects.filter(realm=realm)))
staged_messages = [
{'sender': starr,
'content': "Hey @**Bel Fisher**, check out Zulip's Markdown formatting! "
"You can have:\n* bulleted lists\n * with sub-bullets too\n"
"* **bold**, *italic*, and ~~strikethrough~~ text\n"
"* LaTeX for mathematical formulas, both inline -- $$O(n^2)$$ -- and displayed:\n"
"```math\n\\int_a^b f(t)\, dt=F(b)-F(a)\n```"},
{'sender': fisher,
'content': "My favorite is the syntax highlighting for code blocks\n"
"```python\ndef fib(n: int) -> int:\n # returns the n-th Fibonacci number\n"
" return fib(n-1) + fib(n-2)\n```"},
{'sender': starr,
'content': "I think you forgot your base case there, Bel :laughing:\n"
"```quote\n```python\ndef fib(n: int) -> int:\n # returns the n-th Fibonacci number\n"
" return fib(n-1) + fib(n-2)\n```\n```"},
{'sender': fisher,
'content': "I'm also a big fan of inline link, tweet, video, and image previews. "
"Check out this picture of Çet Whalin[](/static/images/features/whale.png)!"},
{'sender': starr,
'content': "I just set up a custom linkifier, "
"so `#1234` becomes [#1234](github.com/zulip/zulip/1234), "
"a link to the corresponding GitHub issue."},
{'sender': twitter_bot,
'content': 'https://twitter.com/gvanrossum/status/786661035637772288'},
{'sender': fisher,
'content': "Oops, the Twitter bot I set up shouldn't be posting here. Let me go fix that."},
] # type: List[Dict[str, Any]]
messages = [internal_prep_stream_message(
realm, message['sender'], stream.name, 'message formatting', message['content']
) for message in staged_messages]
message_ids = do_send_messages(messages)
preview_message = Message.objects.get(id__in=message_ids, content__icontains='image previews')
do_add_reaction_legacy(starr, preview_message, 'whale')
twitter_message = Message.objects.get(id__in=message_ids, content__icontains='gvanrossum')
# Setting up a twitter integration in dev is a decent amount of work. If you need
# to update this tweet, either copy the format below, or send the link to the tweet
# to chat.zulip.org and ask an admin of that server to get you the rendered_content.
twitter_message.rendered_content = (
'<p><a>https://twitter.com/gvanrossum/status/786661035637772288</a></p>\n'
'<div class="inline-preview-twitter"><div class="twitter-tweet">'
'<a><img class="twitter-avatar" '
'src="https://pbs.twimg.com/profile_images/424495004/GuidoAvatar_bigger.jpg"></a>'
'<p>Great blog post about Zulip\'s use of mypy: '
'<a>http://blog.zulip.org/2016/10/13/static-types-in-python-oh-mypy/</a></p>'
'<span>- Guido van Rossum (@gvanrossum)</span></div></div>')
twitter_message.save(update_fields=['rendered_content'])
# Put a short pause between the whale reaction and this, so that the
# thumbs_up shows up second
do_add_reaction_legacy(starr, preview_message, 'thumbs_up')
开发者ID:gnprice,项目名称:zulip,代码行数:69,代码来源:add_mock_conversation.py
注:本文中的zerver.lib.actions.create_stream_if_needed函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论