# Path: tests/pgshovel/streams/fixtures.py
# DEFAULT_PUBLISHER = uuid.uuid1().bytes
# def reserialize(message):
# def copy(message, **replacements):
# def make_message(payload, sequence=1, publisher=DEFAULT_PUBLISHER):
# def make_messages(payloads, publisher=DEFAULT_PUBLISHER):
# def make_batch_messages(batch_identifier, payloads, **kwargs):
# def transactions():
# def message():


import pytest
from itertools import islice
from kafka import (
    KafkaClient,
    SimpleProducer,
)
from tests.pgshovel.fixtures import (
    cluster,
    create_temporary_database,
)
from tests.pgshovel.streams.fixtures import (
    DEFAULT_PUBLISHER,
    begin,
    transaction,
    transactions,
)
from pgshovel.interfaces.common_pb2 import Snapshot
from pgshovel.interfaces.configurations_pb2 import ReplicationSetConfiguration
from pgshovel.interfaces.replication_pb2 import (
    ConsumerState,
    State,
    BootstrapState,
    TransactionState,
)
from pgshovel.interfaces.streams_pb2 import (
    Header,
    Message,
)
from pgshovel.replication.streams.kafka import KafkaStream
from pgshovel.replication.validation.consumers import SequencingError
from pgshovel.replication.validation.transactions import InvalidEventError
from pgshovel.relay.streams.kafka import KafkaWriter
from pgshovel.streams.utilities import UnableToPrimeError
@pytest.yield_fixture
def sliced_transaction():
    two_transactions = list(islice(transactions(), 6))

    head, remainder = two_transactions[0], two_transactions[1:]
    assert head.batch_operation.begin_operation == begin
    yield remainder



def test_starts_at_beginning_of_stream_for_bootstrapped_state(writer, stream, state):
    writer.push(transaction)
    consumed = list(islice(stream.consume(state), 3))
    assert [message for _, _, message in consumed] == transaction

def test_yields_new_update_state_after_each_message(writer, stream, state):
    expected_states = {
        0: 'in_transaction',
        1: 'in_transaction',
        2: 'committed'
    }

    writer.push(transaction)

    for state, offset, message in islice(stream.consume(state), 3):
        assert state.stream_state.consumer_state.offset == offset
        assert state.stream_state.consumer_state.header == message.header
        assert state.stream_state.transaction_state.WhichOneof('state') == expected_states[offset]

def test_uses_existing_stream_state_if_it_exists(writer, stream, state):

# Path: tests/pgshovel/streams/fixtures.py
# DEFAULT_PUBLISHER = uuid.uuid1().bytes
# def reserialize(message):
# def copy(message, **replacements):
# def make_message(payload, sequence=1, publisher=DEFAULT_PUBLISHER):
# def make_messages(payloads, publisher=DEFAULT_PUBLISHER):
# def make_batch_messages(batch_identifier, payloads, **kwargs):
# def transactions():
# def message():


# Path: tests/pgshovel/replication/streams/kafka.py
import pytest
from itertools import islice
from kafka import (
    KafkaClient,
    SimpleProducer,
)
from tests.pgshovel.fixtures import (
    cluster,
    create_temporary_database,
)
from tests.pgshovel.streams.fixtures import (
    DEFAULT_PUBLISHER,
    begin,
    transaction,
    transactions,
)
from pgshovel.interfaces.common_pb2 import Snapshot
from pgshovel.interfaces.configurations_pb2 import ReplicationSetConfiguration
from pgshovel.interfaces.replication_pb2 import (
    ConsumerState,
    State,
    BootstrapState,
    TransactionState,
)
from pgshovel.interfaces.streams_pb2 import (
    Header,
    Message,
)
from pgshovel.replication.streams.kafka import KafkaStream
from pgshovel.replication.validation.consumers import SequencingError
from pgshovel.replication.validation.transactions import InvalidEventError
from pgshovel.relay.streams.kafka import KafkaWriter
from pgshovel.streams.utilities import UnableToPrimeError

@pytest.yield_fixture
def sliced_transaction():
    two_transactions = list(islice(transactions(), 6))

    head, remainder = two_transactions[0], two_transactions[1:]
    assert head.batch_operation.begin_operation == begin
    yield remainder



def test_starts_at_beginning_of_stream_for_bootstrapped_state(writer, stream, state):
    writer.push(transaction)
    consumed = list(islice(stream.consume(state), 3))
    assert [message for _, _, message in consumed] == transaction

def test_yields_new_update_state_after_each_message(writer, stream, state):
    expected_states = {
        0: 'in_transaction',
        1: 'in_transaction',
        2: 'committed'
    }

    writer.push(transaction)

    for state, offset, message in islice(stream.consume(state), 3):
        assert state.stream_state.consumer_state.offset == offset
        assert state.stream_state.consumer_state.header == message.header
        assert state.stream_state.transaction_state.WhichOneof('state') == expected_states[offset]

def test_uses_existing_stream_state_if_it_exists(writer, stream, state):

    writer.push(islice(transactions(), 6))