<commit_before># coding=utf-8
"""
A simple smoke-test for Google Shared Status.
See: http://code.google.com/apis/talk/jep_extensions/shared_status.html
"""
from gabbletest import (
    exec_test, XmppXmlStream, acknowledge_iq, send_error_reply,
    disconnect_conn, elem, elem_iq
)
from servicetest import (
    EventPattern, assertEquals, assertNotEquals, assertContains,
    assertDoesNotContain
)
import ns
import copy
import constants as cs
import dbus
from twisted.words.xish import xpath, domish
from invisible_helper import SharedStatusStream

presence_types = {'available' : cs.PRESENCE_AVAILABLE,
                  'away'      : cs.PRESENCE_AWAY,
                  'xa'        : cs.PRESENCE_EXTENDED_AWAY,
                  'hidden'    : cs.PRESENCE_HIDDEN,
                  'dnd'       : cs.PRESENCE_BUSY}

def _show_to_shared_status_show(show):
    # Away and extended away don't use shared status.
    assert show not in ('away', 'xa')

    shared_show = 'default'
    if show == 'dnd':
        shared_show = 'dnd'

    shared_invisible = 'false'
    if show == 'hidden':
        shared_invisible = 'true'

    return shared_show, shared_invisible

def _test_remote_status(q, stream, msg, show, list_attrs):
    list_attrs['status'] = list_attrs.get('status') or msg
    stream.set_shared_status_lists(**list_attrs)

    q.expect('stream-iq', iq_type='result')

    q.expect_many(
        EventPattern('dbus-signal', signal='PresenceUpdate',
                     interface=cs.CONN_IFACE_PRESENCE,
                     args=[{1: (0, {show: {'message': msg}})}]),
        EventPattern('dbus-signal', signal='PresencesChanged',
                     interface=cs.CONN_IFACE_SIMPLE_PRESENCE,
                     args=[{1: (presence_types[show], show, msg)}]))

def _test_local_status(q, conn, stream, msg, show, expected_show=None):
    expected_show = expected_show or show
    away = expected_show in ('away', 'xa')

    # Away and extended away are mapped to idle, that is per connection.
    # This means we use <presence/> instead of shared presence.
    if not away:
        wrong_presence_pattern = EventPattern('stream-presence')
    else:
        wrong_presence_pattern = EventPattern('stream-iq',
                                              query_ns=ns.GOOGLE_SHARED_STATUS,
                                              iq_type='set')
    q.forbid_events([wrong_presence_pattern])

    conn.SimplePresence.SetPresence(show, msg)

    max_status_message_length = int(stream.max_status_message_length)

    if not away:
        event = q.expect('stream-iq', query_ns=ns.GOOGLE_SHARED_STATUS,
                                     iq_type='set')

        shared_show, shared_invisible = _show_to_shared_status_show(expected_show)

        _status = xpath.queryForNodes('//status', event.query)[0]
        assertEquals(msg[:max_status_message_length], _status.children[0])
        _show = xpath.queryForNodes('//show', event.query)[0]
        assertEquals(shared_show, _show.children[0])
        _invisible = xpath.queryForNodes('//invisible', event.query)[0]
        assertEquals(shared_invisible, _invisible.getAttribute('value'))
    else:
        q.expect('stream-presence')

    q.expect_many(
        EventPattern('dbus-signal', signal='PresenceUpdate',
                     interface=cs.CONN_IFACE_PRESENCE,
                     args=[{1: (0, {expected_show: {'message':
                            msg[:max_status_message_length]}})}]),
        EventPattern('dbus-signal', signal='PresencesChanged',
                     interface=cs.CONN_IFACE_SIMPLE_PRESENCE,
                     args=[{1: (presence_types[expected_show], expected_show,
                            msg[:max_status_message_length])}]))

    q.unforbid_events([wrong_presence_pattern])

def test(q, bus, conn, stream):
    q.expect_many(EventPattern('stream-iq', query_ns=ns.GOOGLE_SHARED_STATUS,
                               iq_type='get'),
                  EventPattern('stream-iq', query_ns=ns.GOOGLE_SHARED_STATUS,
                               iq_type='set'),
                  EventPattern('stream-presence'))

    # Set shared status to dnd.
    _test_local_status(q, conn, stream, "Don't disturb, buddy.", "dnd")

    # Test maximum status message length
    max_status_message_length = int(stream.max_status_message_length)
    _test_local_status(q, conn, stream, "ab" * max_status_message_length, "dnd")

    # Test invisibility
    _test_local_status(q, conn, stream, "Peekabo", "hidden")

    # Set shared status to default, local status to away.
    _test_local_status(q, conn, stream, "I'm away right now", "away")
    _test_local_status(q, conn, stream, "cd" * max_status_message_length, "away")

    # Test interaction with another client.
    _test_local_status(q, conn, stream, "Don't disturb, buddy.", "dnd")
    _test_remote_status(q, stream, "This is me busy, set from another client.",
                        "dnd", {"show" : "dnd"})
    _test_remote_status(q, stream, "This is me available, set from another client.",
                        "available", {"show" : "default"})

    # Change min version
    stream.set_shared_status_lists(min_version="1")
    q.expect('stream-iq', iq_type='result')

    # Going invisible now should fail, we should just be dnd.
    _test_local_status(q, conn, stream, "Peekabo", "hidden", "dnd")

    # Let's go back to version 2
    stream.set_shared_status_lists(min_version="2")
    q.expect('stream-iq', iq_type='result')

    # "hidden" should work again.
    _test_local_status(q, conn, stream, "Peekabo", "hidden")

    # Changing min version mid-flight should make us 'dnd'
    stream.set_shared_status_lists(min_version="1")
    q.expect('stream-iq', iq_type='result')

    q.expect_many(
        EventPattern('dbus-signal', signal='PresenceUpdate',
                     interface=cs.CONN_IFACE_PRESENCE,
                     args=[{1: (0, {'dnd': {'message': "Peekabo"}})}]),
        EventPattern('dbus-signal', signal='PresencesChanged',
                     interface=cs.CONN_IFACE_SIMPLE_PRESENCE,
                     args=[{1: (cs.PRESENCE_BUSY, 'dnd', "Peekabo")}]))

def _test_on_connect(q, bus, conn, stream, shared_status, show, msg, expected_show=None):
    expected_show = expected_show or show
    _status, _show, _invisible = shared_status
    stream.shared_status = shared_status

    forbidden_even_patterns = [EventPattern('stream-presence'),
                               EventPattern('stream-iq', query_ns=ns.PRIVACY,
                                            iq_type='get')]
    q.forbid_events(forbidden_even_patterns)

    conn.SimplePresence.SetPresence(show, msg)
    conn.Connect()

    _, event = q.expect_many(EventPattern('stream-iq', query_ns=ns.GOOGLE_SHARED_STATUS,
                                          iq_type='get'),
                             EventPattern('stream-iq', query_ns=ns.GOOGLE_SHARED_STATUS,
                                          iq_type='set'))

    shared_show, shared_invisible = _show_to_shared_status_show(show)

    _status = xpath.queryForNodes('//status', event.query)[0]
    assertEquals(msg, _status.children[0])
    _show = xpath.queryForNodes('//show', event.query)[0]
    assertEquals(shared_show, _show.children[0])
    _invisible = xpath.queryForNodes('//invisible', event.query)[0]
    assertEquals(shared_invisible, _invisible.getAttribute('value'))

    q.expect_many(
        EventPattern('dbus-signal', signal='PresenceUpdate',
                     interface=cs.CONN_IFACE_PRESENCE,
                     args=[{1: (0, {expected_show: {'message': msg}})}]),
        EventPattern('dbus-signal', signal='PresencesChanged',
                     interface=cs.CONN_IFACE_SIMPLE_PRESENCE,
                     args=[{1: (presence_types[expected_show],
                                expected_show, msg)}]),
        EventPattern('dbus-signal', signal='StatusChanged',
                     args=[cs.CONN_STATUS_CONNECTED, cs.CSR_REQUESTED]))

    q.unforbid_events(forbidden_even_patterns)

def test_connect_available(q, bus, conn, stream):
    _test_on_connect(q, bus, conn, stream,  ("I'm busy, buddy.", 'dnd', 'false'),
                     'available', "I'm here, baby.")

def test_connect_chat(q, bus, conn, stream):
    _test_on_connect(q, bus, conn, stream,  ("I'm busy, buddy.", 'dnd', 'false'),
                     'chat', "Do you want to chat?", 'available')

def test_connect_dnd(q, bus, conn, stream):
    _test_on_connect(q, bus, conn, stream,  ("Chat with me.", 'default', 'false'),
                     'dnd', "I'm busy, buddy.")

def test_connect_hidden(q, bus, conn, stream):
    _test_on_connect(q, bus, conn, stream,  ("Chat with me.", 'default', 'false'),
                     'hidden', "I see, but I can't be seen")

def test_connect_hidden_not_available(q, bus, conn, stream):<commit_msg>presence/shared-status.py: fix a typo (even → event)<commit_after># coding=utf-8
"""
A simple smoke-test for Google Shared Status.
See: http://code.google.com/apis/talk/jep_extensions/shared_status.html
"""
from gabbletest import (
    exec_test, XmppXmlStream, acknowledge_iq, send_error_reply,
    disconnect_conn, elem, elem_iq
)
from servicetest import (
    EventPattern, assertEquals, assertNotEquals, assertContains,
    assertDoesNotContain
)
import ns
import copy
import constants as cs
import dbus
from twisted.words.xish import xpath, domish
from invisible_helper import SharedStatusStream

presence_types = {'available' : cs.PRESENCE_AVAILABLE,
                  'away'      : cs.PRESENCE_AWAY,
                  'xa'        : cs.PRESENCE_EXTENDED_AWAY,
                  'hidden'    : cs.PRESENCE_HIDDEN,
                  'dnd'       : cs.PRESENCE_BUSY}

def _show_to_shared_status_show(show):
    # Away and extended away don't use shared status.
    assert show not in ('away', 'xa')

    shared_show = 'default'
    if show == 'dnd':
        shared_show = 'dnd'

    shared_invisible = 'false'
    if show == 'hidden':
        shared_invisible = 'true'

    return shared_show, shared_invisible

def _test_remote_status(q, stream, msg, show, list_attrs):
    list_attrs['status'] = list_attrs.get('status') or msg
    stream.set_shared_status_lists(**list_attrs)

    q.expect('stream-iq', iq_type='result')

    q.expect_many(
        EventPattern('dbus-signal', signal='PresenceUpdate',
                     interface=cs.CONN_IFACE_PRESENCE,
                     args=[{1: (0, {show: {'message': msg}})}]),
        EventPattern('dbus-signal', signal='PresencesChanged',
                     interface=cs.CONN_IFACE_SIMPLE_PRESENCE,
                     args=[{1: (presence_types[show], show, msg)}]))

def _test_local_status(q, conn, stream, msg, show, expected_show=None):
    expected_show = expected_show or show
    away = expected_show in ('away', 'xa')

    # Away and extended away are mapped to idle, that is per connection.
    # This means we use <presence/> instead of shared presence.
    if not away:
        wrong_presence_pattern = EventPattern('stream-presence')
    else:
        wrong_presence_pattern = EventPattern('stream-iq',
                                              query_ns=ns.GOOGLE_SHARED_STATUS,
                                              iq_type='set')
    q.forbid_events([wrong_presence_pattern])

    conn.SimplePresence.SetPresence(show, msg)

    max_status_message_length = int(stream.max_status_message_length)

    if not away:
        event = q.expect('stream-iq', query_ns=ns.GOOGLE_SHARED_STATUS,
                                     iq_type='set')

        shared_show, shared_invisible = _show_to_shared_status_show(expected_show)

        _status = xpath.queryForNodes('//status', event.query)[0]
        assertEquals(msg[:max_status_message_length], _status.children[0])
        _show = xpath.queryForNodes('//show', event.query)[0]
        assertEquals(shared_show, _show.children[0])
        _invisible = xpath.queryForNodes('//invisible', event.query)[0]
        assertEquals(shared_invisible, _invisible.getAttribute('value'))
    else:
        q.expect('stream-presence')

    q.expect_many(
        EventPattern('dbus-signal', signal='PresenceUpdate',
                     interface=cs.CONN_IFACE_PRESENCE,
                     args=[{1: (0, {expected_show: {'message':
                            msg[:max_status_message_length]}})}]),
        EventPattern('dbus-signal', signal='PresencesChanged',
                     interface=cs.CONN_IFACE_SIMPLE_PRESENCE,
                     args=[{1: (presence_types[expected_show], expected_show,
                            msg[:max_status_message_length])}]))

    q.unforbid_events([wrong_presence_pattern])

def test(q, bus, conn, stream):
    q.expect_many(EventPattern('stream-iq', query_ns=ns.GOOGLE_SHARED_STATUS,
                               iq_type='get'),
                  EventPattern('stream-iq', query_ns=ns.GOOGLE_SHARED_STATUS,
                               iq_type='set'),
                  EventPattern('stream-presence'))

    # Set shared status to dnd.
    _test_local_status(q, conn, stream, "Don't disturb, buddy.", "dnd")

    # Test maximum status message length
    max_status_message_length = int(stream.max_status_message_length)
    _test_local_status(q, conn, stream, "ab" * max_status_message_length, "dnd")

    # Test invisibility
    _test_local_status(q, conn, stream, "Peekabo", "hidden")

    # Set shared status to default, local status to away.
    _test_local_status(q, conn, stream, "I'm away right now", "away")
    _test_local_status(q, conn, stream, "cd" * max_status_message_length, "away")

    # Test interaction with another client.
    _test_local_status(q, conn, stream, "Don't disturb, buddy.", "dnd")
    _test_remote_status(q, stream, "This is me busy, set from another client.",
                        "dnd", {"show" : "dnd"})
    _test_remote_status(q, stream, "This is me available, set from another client.",
                        "available", {"show" : "default"})

    # Change min version
    stream.set_shared_status_lists(min_version="1")
    q.expect('stream-iq', iq_type='result')

    # Going invisible now should fail, we should just be dnd.
    _test_local_status(q, conn, stream, "Peekabo", "hidden", "dnd")

    # Let's go back to version 2
    stream.set_shared_status_lists(min_version="2")
    q.expect('stream-iq', iq_type='result')

    # "hidden" should work again.
    _test_local_status(q, conn, stream, "Peekabo", "hidden")

    # Changing min version mid-flight should make us 'dnd'
    stream.set_shared_status_lists(min_version="1")
    q.expect('stream-iq', iq_type='result')

    q.expect_many(
        EventPattern('dbus-signal', signal='PresenceUpdate',
                     interface=cs.CONN_IFACE_PRESENCE,
                     args=[{1: (0, {'dnd': {'message': "Peekabo"}})}]),
        EventPattern('dbus-signal', signal='PresencesChanged',
                     interface=cs.CONN_IFACE_SIMPLE_PRESENCE,
                     args=[{1: (cs.PRESENCE_BUSY, 'dnd', "Peekabo")}]))

def _test_on_connect(q, bus, conn, stream, shared_status, show, msg, expected_show=None):
    expected_show = expected_show or show
    _status, _show, _invisible = shared_status
    stream.shared_status = shared_status

    forbidden_event_patterns = [EventPattern('stream-presence'),
                                EventPattern('stream-iq', query_ns=ns.PRIVACY,
                                             iq_type='get')]
    q.forbid_events(forbidden_event_patterns)

    conn.SimplePresence.SetPresence(show, msg)
    conn.Connect()

    _, event = q.expect_many(EventPattern('stream-iq', query_ns=ns.GOOGLE_SHARED_STATUS,
                                          iq_type='get'),
                             EventPattern('stream-iq', query_ns=ns.GOOGLE_SHARED_STATUS,
                                          iq_type='set'))

    shared_show, shared_invisible = _show_to_shared_status_show(show)

    _status = xpath.queryForNodes('//status', event.query)[0]
    assertEquals(msg, _status.children[0])
    _show = xpath.queryForNodes('//show', event.query)[0]
    assertEquals(shared_show, _show.children[0])
    _invisible = xpath.queryForNodes('//invisible', event.query)[0]
    assertEquals(shared_invisible, _invisible.getAttribute('value'))

    q.expect_many(
        EventPattern('dbus-signal', signal='PresenceUpdate',
                     interface=cs.CONN_IFACE_PRESENCE,
                     args=[{1: (0, {expected_show: {'message': msg}})}]),
        EventPattern('dbus-signal', signal='PresencesChanged',
                     interface=cs.CONN_IFACE_SIMPLE_PRESENCE,
                     args=[{1: (presence_types[expected_show],
                                expected_show, msg)}]),
        EventPattern('dbus-signal', signal='StatusChanged',
                     args=[cs.CONN_STATUS_CONNECTED, cs.CSR_REQUESTED]))

    q.unforbid_events(forbidden_event_patterns)

def test_connect_available(q, bus, conn, stream):
    _test_on_connect(q, bus, conn, stream,  ("I'm busy, buddy.", 'dnd', 'false'),
                     'available', "I'm here, baby.")

def test_connect_chat(q, bus, conn, stream):
    _test_on_connect(q, bus, conn, stream,  ("I'm busy, buddy.", 'dnd', 'false'),
                     'chat', "Do you want to chat?", 'available')

def test_connect_dnd(q, bus, conn, stream):
    _test_on_connect(q, bus, conn, stream,  ("Chat with me.", 'default', 'false'),
                     'dnd', "I'm busy, buddy.")

def test_connect_hidden(q, bus, conn, stream):
    _test_on_connect(q, bus, conn, stream,  ("Chat with me.", 'default', 'false'),
                     'hidden', "I see, but I can't be seen")

def test_connect_hidden_not_available(q, bus, conn, stream):
{'fraction_non_alphanumeric': 0.1220681100586378, 'fraction_numerical': 0.003044654939106901, 'mean_word_length': 2.745143581081081, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 22, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6526809', 'n_tokens_mistral': 5577, 'n_tokens_neox': 5346, 'n_words': 1099}