# 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
import psycopg2
import uuid
from copy import deepcopy
from itertools import islice
from kafka import (
    KafkaClient,
    SimpleProducer,
)
from kazoo.client import KazooClient
from kafka import KafkaClient
from tests.pgshovel.fixtures import (
    cluster,
    create_temporary_database
)
from tests.pgshovel.streams.fixtures import (
    batch_identifier,
    make_batch_messages,
    transaction,
)
from pgshovel.administration import (
    create_set,
    get_managed_databases,
    initialize_cluster,
)
from pgshovel.cluster import Cluster
from pgshovel.interfaces.common_pb2 import (
    Column,
    Row,
    Snapshot,
    Tick,
    Timestamp,
)
from pgshovel.interfaces.configurations_pb2 import ReplicationSetConfiguration
from pgshovel.interfaces.streams_pb2 import (
    BeginOperation,
    CommitOperation,
    MutationOperation,
    RollbackOperation,
)
from pgshovel.relay.streams.kafka import KafkaWriter
from pgshovel.replication.loaders.simple import SimpleLoader
from pgshovel.replication.streams.kafka import KafkaStream
from pgshovel.replication.targets.postgresql import PostgreSQLTarget
from pgshovel.utilities.conversions import to_snapshot

    def limited_to(self, limit):
        self.limit = limit
        return self

    def consume(self, *args, **kwargs):
        iterator = super(LimitedKafkaStream, self).consume(*args, **kwargs)
        return islice(iterator, self.limit)


def get_mutation(transaction_id=1, user_1_name='kevin bacon'):
    return MutationOperation(
        id=1,
        schema='public',
        table='auth_user',
        operation=MutationOperation.UPDATE,
        identity_columns=['id'],
        new=Row(
            columns=[
                Column(name='id', integer64=1),
                Column(name='username', string=user_1_name),
            ],
        ),
        old=Row(
            columns=[
                Column(name='id', integer64=1),
                Column(name='username', string='example2'),
            ],
        ),
        timestamp=Timestamp(seconds=0, nanos=0),

# 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/targets/postgres.py
import pytest
import psycopg2
import uuid
from copy import deepcopy
from itertools import islice
from kafka import (
    KafkaClient,
    SimpleProducer,
)
from kazoo.client import KazooClient
from kafka import KafkaClient
from tests.pgshovel.fixtures import (
    cluster,
    create_temporary_database
)
from tests.pgshovel.streams.fixtures import (
    batch_identifier,
    make_batch_messages,
    transaction,
)
from pgshovel.administration import (
    create_set,
    get_managed_databases,
    initialize_cluster,
)
from pgshovel.cluster import Cluster
from pgshovel.interfaces.common_pb2 import (
    Column,
    Row,
    Snapshot,
    Tick,
    Timestamp,
)
from pgshovel.interfaces.configurations_pb2 import ReplicationSetConfiguration
from pgshovel.interfaces.streams_pb2 import (
    BeginOperation,
    CommitOperation,
    MutationOperation,
    RollbackOperation,
)
from pgshovel.relay.streams.kafka import KafkaWriter
from pgshovel.replication.loaders.simple import SimpleLoader
from pgshovel.replication.streams.kafka import KafkaStream
from pgshovel.replication.targets.postgresql import PostgreSQLTarget
from pgshovel.utilities.conversions import to_snapshot


    def limited_to(self, limit):
        self.limit = limit
        return self

    def consume(self, *args, **kwargs):
        iterator = super(LimitedKafkaStream, self).consume(*args, **kwargs)
        return islice(iterator, self.limit)


def get_mutation(transaction_id=1, user_1_name='kevin bacon'):
    return MutationOperation(
        id=1,
        schema='public',
        table='auth_user',
        operation=MutationOperation.UPDATE,
        identity_columns=['id'],
        new=Row(
            columns=[
                Column(name='id', integer64=1),
                Column(name='username', string=user_1_name),
            ],
        ),
        old=Row(
            columns=[
                Column(name='id', integer64=1),
                Column(name='username', string='example2'),
            ],
        ),
        timestamp=Timestamp(seconds=0, nanos=0),

        transaction=transaction_id,