import asyncio

from kafka.admin import NewTopic, NewPartitions
from kafka.admin.config_resource import ConfigResource, ConfigResourceType

from aiokafka.admin import AIOKafkaAdminClient
from aiokafka.consumer import AIOKafkaConsumer
from aiokafka.producer import AIOKafkaProducer
from aiokafka.structs import TopicPartition
from ._testutil import (
    KafkaIntegrationTestCase, kafka_versions, run_until_complete
)


class TestAdmin(KafkaIntegrationTestCase):
    async def create_admin(self):
        admin = AIOKafkaAdminClient(bootstrap_servers=self.hosts)
        await admin.start()
        self.add_cleanup(admin.close)
        return admin

    @kafka_versions('>=0.10.0.0')
    @run_until_complete
    async def test_metadata(self):
        admin = await self.create_admin()
        metadata = await admin._get_cluster_metadata()
        assert metadata.brokers is not None
        assert metadata.topics is not None
        assert len(metadata.brokers) == 1

    @kafka_versions('>=0.10.1.0')
    @run_until_complete
    async def test_create_topics(self):
        admin = await self.create_admin()
        resp = await admin.create_topics([NewTopic(self.topic, 1, 1)])