# 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 configuration():
    yield {'hosts': 'kafka:9092'}


@pytest.yield_fixture
def stream(configuration, cluster, client):
    stream = KafkaStream.configure(configuration, cluster, 'default')
    client.ensure_topic_exists(stream.topic)
    yield stream


@pytest.yield_fixture
def client(configuration):
    yield KafkaClient(configuration['hosts'])


@pytest.yield_fixture
def writer(client, stream):
    producer = SimpleProducer(client)
    yield KafkaWriter(producer, stream.topic)


@pytest.yield_fixture
def 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 configuration():
    yield {'hosts': 'kafka:9092'}


@pytest.yield_fixture
def stream(configuration, cluster, client):
    stream = KafkaStream.configure(configuration, cluster, 'default')
    client.ensure_topic_exists(stream.topic)
    yield stream


@pytest.yield_fixture
def client(configuration):
    yield KafkaClient(configuration['hosts'])


@pytest.yield_fixture
def writer(client, stream):
    producer = SimpleProducer(client)
    yield KafkaWriter(producer, stream.topic)


@pytest.yield_fixture
def state():

    bootstrap_state = BootstrapState(